Skip to content

Skills API

Utility skills for asset processing and spatial computing tasks.

embodied_gen.skills.asset-process

Asset processing skills for EmbodiedGen.

__getattr__

__getattr__(name: str) -> object

Lazily expose asset processing APIs without import side effects.

Source code in embodied_gen/skills/asset-process/__init__.py
29
30
31
32
33
34
35
36
def __getattr__(name: str) -> object:
    """Lazily expose asset processing APIs without import side effects."""
    if name not in __all__:
        msg = f"module {__name__!r} has no attribute {name!r}"
        raise AttributeError(msg)

    module = import_module(f"{__name__}.asset_process")
    return getattr(module, name)

embodied_gen.skills.asset-process.asset_process

Asset processing utilities for scaling and rotating 3D assets.

AssetProcessConfig dataclass

AssetProcessConfig(urdf_path: str, scale_factor: float = 1.0, rot_xyz: tuple[float, float, float] = (0.0, 0.0, 0.0), keep_urdf_raw_rot: bool = False, inplace: bool = False, output_dir: Optional[str] = None)

Configuration for asset scaling and rotation.

Parameters:

Name Type Description Default
urdf_path str

Path to the URDF file to process.

required
scale_factor float

Positive uniform scaling factor.

1.0
rot_xyz tuple[float, float, float]

XYZ Euler rotation in degrees.

(0.0, 0.0, 0.0)
keep_urdf_raw_rot bool

Whether to preserve the original URDF rotation.

False
inplace bool

Whether to modify the source asset directly.

False
output_dir Optional[str]

Target asset directory used when inplace is false.

None

AssetProcessor

AssetProcessor(urdf_path: str | Path, scale_factor: float = 1.0, rot_xyz: tuple[float, float, float] = (0.0, 0.0, 0.0), keep_urdf_raw_rot: bool = False, output_dir: Optional[str | Path] = None, inplace: bool = False)

Scale and rotate mesh, collision, and Gaussian-splat asset files.

Initialize an asset processor.

Parameters:

Name Type Description Default
urdf_path str | Path

Path to the URDF file to process.

required
scale_factor float

Positive uniform scaling factor.

1.0
rot_xyz tuple[float, float, float]

XYZ Euler rotation in degrees.

(0.0, 0.0, 0.0)
keep_urdf_raw_rot bool

Whether to preserve the original visual and collision rotation after baking the asset rotation.

False
output_dir Optional[str | Path]

Target asset directory used when inplace is false.

None
inplace bool

Whether to modify the source asset directly.

False

Raises:

Type Description
FileNotFoundError

If the URDF file does not exist.

ValueError

If the transform or output configuration is invalid.

Source code in embodied_gen/skills/asset-process/asset_process.py
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def __init__(
    self,
    urdf_path: str | Path,
    scale_factor: float = 1.0,
    rot_xyz: tuple[float, float, float] = (0.0, 0.0, 0.0),
    keep_urdf_raw_rot: bool = False,
    output_dir: Optional[str | Path] = None,
    inplace: bool = False,
) -> None:
    """Initialize an asset processor.

    Args:
        urdf_path: Path to the URDF file to process.
        scale_factor: Positive uniform scaling factor.
        rot_xyz: XYZ Euler rotation in degrees.
        keep_urdf_raw_rot: Whether to preserve the original visual and
            collision rotation after baking the asset rotation.
        output_dir: Target asset directory used when ``inplace`` is false.
        inplace: Whether to modify the source asset directly.

    Raises:
        FileNotFoundError: If the URDF file does not exist.
        ValueError: If the transform or output configuration is invalid.
    """
    self.urdf_path = Path(urdf_path)
    self.scale_factor = float(scale_factor)
    self.rot_xyz = tuple(float(value) for value in rot_xyz)
    self.keep_urdf_raw_rot = keep_urdf_raw_rot
    self.inplace = inplace

    if not self.urdf_path.is_file():
        raise FileNotFoundError(f"URDF file not found: {self.urdf_path}")
    if not math.isfinite(self.scale_factor) or self.scale_factor <= 0:
        raise ValueError(
            f"Scale factor must be positive and finite, got: "
            f"{self.scale_factor}"
        )
    if len(self.rot_xyz) != 3 or not all(
        math.isfinite(value) for value in self.rot_xyz
    ):
        raise ValueError(
            "rot_xyz must contain three finite degree values, got: "
            f"{self.rot_xyz}"
        )

    if self.urdf_path.parent.name == URDF_RESULT_DIR:
        self.asset_dir = self.urdf_path.parent.parent
        self.result_dir = Path(URDF_RESULT_DIR)
    else:
        self.asset_dir = self.urdf_path.parent
        self.result_dir = Path()
    self.node_name = self.urdf_path.stem
    self.rotation = Rotation.from_euler("xyz", self.rot_xyz, degrees=True)
    self.rotation_transform = np.eye(4)
    self.rotation_transform[:3, :3] = self.rotation.as_matrix()

    if self.inplace:
        self.output_dir = self.asset_dir.parent
    else:
        if output_dir is None:
            raise ValueError("output_dir is required when inplace=False")
        self.output_dir = Path(output_dir)
        if self.output_dir.resolve() == self.asset_dir.resolve():
            raise ValueError(
                "output_dir must differ from the source asset directory; "
                "use inplace=True to modify the source asset"
            )
process
process() -> Path

Run the complete asset processing workflow.

Returns:

Type Description
Path

Path to the processed URDF file.

Source code in embodied_gen/skills/asset-process/asset_process.py
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def process(self) -> Path:
    """Run the complete asset processing workflow.

    Returns:
        Path to the processed URDF file.
    """
    if self.inplace:
        output_asset_dir = self.asset_dir
        output_urdf_path = self.urdf_path
    else:
        output_asset_dir = self.output_dir
        output_urdf_path = self._copy_asset_structure(output_asset_dir)

    self._process_asset_files_parallel(output_asset_dir)
    actual_height = self._calculate_actual_height(output_asset_dir)
    self._write_urdf_height(output_urdf_path, actual_height)
    if self.keep_urdf_raw_rot:
        self._update_urdf_origins(output_urdf_path)

    logger.info(
        "Processed %s with scale=%s, rot_xyz=%s -> %s",
        self.asset_dir,
        self.scale_factor,
        self.rot_xyz,
        output_asset_dir,
    )
    return output_urdf_path

entrypoint

entrypoint() -> None

Run the asset processing CLI.

Source code in embodied_gen/skills/asset-process/asset_process.py
428
429
430
431
432
433
434
435
436
437
438
439
def entrypoint() -> None:
    """Run the asset processing CLI."""
    config = tyro.cli(AssetProcessConfig)
    output_urdf = process_asset(
        urdf_path=config.urdf_path,
        scale_factor=config.scale_factor,
        rot_xyz=config.rot_xyz,
        keep_urdf_raw_rot=config.keep_urdf_raw_rot,
        output_dir=config.output_dir,
        inplace=config.inplace,
    )
    logger.info(f"Processed asset successfully: {output_urdf}")

process_asset

process_asset(urdf_path: str | Path, scale_factor: float = 1.0, rot_xyz: tuple[float, float, float] = (0.0, 0.0, 0.0), keep_urdf_raw_rot: bool = False, output_dir: Optional[str | Path] = None, inplace: bool = False) -> Path

Scale and rotate a complete URDF-based asset.

Source code in embodied_gen/skills/asset-process/asset_process.py
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
def process_asset(
    urdf_path: str | Path,
    scale_factor: float = 1.0,
    rot_xyz: tuple[float, float, float] = (0.0, 0.0, 0.0),
    keep_urdf_raw_rot: bool = False,
    output_dir: Optional[str | Path] = None,
    inplace: bool = False,
) -> Path:
    """Scale and rotate a complete URDF-based asset."""
    processor = AssetProcessor(
        urdf_path=urdf_path,
        scale_factor=scale_factor,
        rot_xyz=rot_xyz,
        keep_urdf_raw_rot=keep_urdf_raw_rot,
        output_dir=output_dir,
        inplace=inplace,
    )
    return processor.process()

embodied_gen.skills.spatial-computing

FloorplanVisualizer

Static utility class for visualizing floorplans.

draw_poly staticmethod
draw_poly(ax: Axes, poly: Geometry, **kwargs) -> None

Draw a polygon or multi-polygon on matplotlib axes.

Parameters:

Name Type Description Default
ax Axes

Matplotlib axes object.

required
poly Geometry

Shapely Polygon or MultiPolygon to draw.

required
**kwargs

Additional arguments passed to ax.fill().

{}
Source code in embodied_gen/skills/spatial-computing/core/visualizer.py
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@staticmethod
def draw_poly(ax: Axes, poly: Geometry, **kwargs) -> None:
    """Draw a polygon or multi-polygon on matplotlib axes.

    Args:
        ax: Matplotlib axes object.
        poly: Shapely Polygon or MultiPolygon to draw.
        **kwargs: Additional arguments passed to ax.fill().

    """
    if poly.is_empty:
        return

    geoms = poly.geoms if hasattr(poly, "geoms") else [poly]

    color = kwargs.pop("color", None)
    if color is None:
        cmap = plt.get_cmap("tab10")
        colors = [cmap(i) for i in range(len(geoms))]
    else:
        colors = [color] * len(geoms)

    for i, p in enumerate(geoms):
        if p.is_empty:
            continue
        x, y = p.exterior.xy
        ax.fill(x, y, facecolor=colors[i], **kwargs)
plot classmethod
plot(rooms: dict[str, Geometry], footprints: dict[str, Geometry], occ_area: Geometry, save_path: str, trajectory: ndarray | None = None, arrow_stride: int = 10, current_index: int | None = None, point_markers: bool = True, dpi: int = 300) -> None

Generate and save a floorplan visualization.

Parameters:

Name Type Description Default
rooms dict[str, Geometry]

Dictionary mapping room names to floor polygons.

required
footprints dict[str, Geometry]

Dictionary mapping object names to footprint polygons.

required
occ_area Geometry

Union of all occupied areas.

required
save_path str

Path to save the output image.

required
trajectory ndarray | None

Optional (N, 2) or (N, 3) array of waypoints. When the third column (rot_deg, tangent heading) is present, heading arrows are drawn. Rendered as a red curve overlay.

None
arrow_stride int

Draw a heading arrow every arrow_stride points (0 disables arrows). Ignored when current_index is set.

10
current_index int | None

Animation frame index. When set, only the traveled path (up to this index) is drawn, with a green dot at the current position and a red heading arrow; the future path is hidden.

None
point_markers bool

When True, mark every trajectory point with a small red dot (in addition to the curve).

True
dpi int

Output image resolution in dots per inch.

300
Source code in embodied_gen/skills/spatial-computing/core/visualizer.py
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@classmethod
def plot(
    cls,
    rooms: dict[str, Geometry],
    footprints: dict[str, Geometry],
    occ_area: Geometry,
    save_path: str,
    trajectory: np.ndarray | None = None,
    arrow_stride: int = 10,
    current_index: int | None = None,
    point_markers: bool = True,
    dpi: int = 300,
) -> None:
    """Generate and save a floorplan visualization.

    Args:
        rooms: Dictionary mapping room names to floor polygons.
        footprints: Dictionary mapping object names to footprint polygons.
        occ_area: Union of all occupied areas.
        save_path: Path to save the output image.
        trajectory: Optional (N, 2) or (N, 3) array of waypoints. When the
            third column (rot_deg, tangent heading) is present, heading
            arrows are drawn. Rendered as a red curve overlay.
        arrow_stride: Draw a heading arrow every ``arrow_stride`` points
            (0 disables arrows). Ignored when ``current_index`` is set.
        current_index: Animation frame index. When set, only the traveled
            path (up to this index) is drawn, with a green dot at the
            current position and a red heading arrow; the future path is
            hidden.
        point_markers: When True, mark every trajectory point with a small
            red dot (in addition to the curve).
        dpi: Output image resolution in dots per inch.

    """
    fig, ax = plt.subplots(figsize=(10, 10))
    ax.set_aspect("equal")
    cmap_rooms = plt.get_cmap("Pastel1")

    cls._draw_room_floors(ax, rooms, cmap_rooms)
    cls._draw_occupied_area(ax, occ_area)
    cls._draw_footprint_outlines(ax, footprints)
    cls._draw_footprint_labels(ax, footprints)
    cls._draw_room_labels(ax, rooms)
    if trajectory is not None and len(trajectory) > 1:
        cls._draw_trajectory(
            ax,
            np.asarray(trajectory),
            arrow_stride,
            current_index,
            point_markers,
        )
    cls._configure_axes(ax, rooms, occ_area)

    ax.set_title("")
    ax.set_xlabel("")
    ax.set_ylabel("")
    ax.set_xticks([])
    ax.set_yticks([])
    for spine in ax.spines.values():
        spine.set_visible(False)
    fig.subplots_adjust(left=0, right=1, top=1, bottom=0)
    fig.patch.set_alpha(0)
    ax.patch.set_alpha(0)
    plt.savefig(
        save_path,
        dpi=dpi,
        bbox_inches="tight",
        pad_inches=0,
        transparent=True,
    )
    plt.close(fig)

UrdfSemanticInfoCollector

UrdfSemanticInfoCollector(mesh_sample_num: int = DEFAULT_MESH_SAMPLE_NUM, ignore_items: list[str] | None = None)

Collector for URDF semantic information.

Parses URDF files to extract room layouts, object footprints, and provides methods for adding new instances and updating URDF/USD files.

Attributes:

Name Type Description
mesh_sample_num

Number of points to sample from meshes.

ignore_items

List of item name patterns to ignore.

instances dict[str, Polygon]

Dictionary of instance name to footprint polygon.

instance_meta dict[str, dict]

Dictionary of instance metadata (mesh path, pose).

rooms dict[str, Geometry]

Dictionary of room polygons.

footprints dict[str, Geometry]

Dictionary of object footprints.

occ_area Geometry

Union of all occupied areas.

floor_union Geometry

Union of all floor polygons.

Initialize the collector.

Parameters:

Name Type Description Default
mesh_sample_num int

Number of points to sample from meshes.

DEFAULT_MESH_SAMPLE_NUM
ignore_items list[str] | None

List of item name patterns to ignore during parsing.

None
Source code in embodied_gen/skills/spatial-computing/core/collector.py
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def __init__(
    self,
    mesh_sample_num: int = DEFAULT_MESH_SAMPLE_NUM,
    ignore_items: list[str] | None = None,
) -> None:
    """Initialize the collector.

    Args:
        mesh_sample_num: Number of points to sample from meshes.
        ignore_items: List of item name patterns to ignore during parsing.

    """
    self.mesh_sample_num = mesh_sample_num
    self.ignore_items = ignore_items or list(DEFAULT_IGNORE_ITEMS)

    self.instances: dict[str, Polygon] = {}
    self.instance_meta: dict[str, dict] = {}
    self.rooms: dict[str, Geometry] = {}
    self.footprints: dict[str, Geometry] = {}
    self.occ_area: Geometry = Polygon()
    self.floor_union: Geometry = Polygon()

    self.urdf_path: str = ""
    self._tree: ET.ElementTree | None = None
    self._root: ET.Element | None = None
add_instance
add_instance(asset_path: str, instance_key: str, in_room: str | None = None, on_instance: str | None = None, beside_instance: str | None = None, beside_distance: float = DEFAULT_BESIDE_DISTANCE, rotation_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY, n_max_attempt: int = DEFAULT_MAX_PLACEMENT_ATTEMPTS, place_strategy: Literal['top', 'random'] = 'random') -> list[float] | None

Add a new instance to the scene with automatic placement.

Parameters:

Name Type Description Default
asset_path str

Path to the asset mesh file.

required
instance_key str

Unique key for the new instance.

required
in_room str | None

Optional room name to constrain placement.

None
on_instance str | None

Optional instance name to place on top of.

None
beside_instance str | None

Optional instance name to place beside (on floor).

None
beside_distance float

Initial buffer distance from the target instance for beside placement (meters). Will auto-expand if needed.

DEFAULT_BESIDE_DISTANCE
rotation_rpy tuple[float, float, float]

Initial rotation in roll-pitch-yaw.

DEFAULT_ROTATION_RPY
n_max_attempt int

Maximum placement attempts.

DEFAULT_MAX_PLACEMENT_ATTEMPTS
place_strategy Literal['top', 'random']

Either "top" or "random".

'random'

Returns:

Type Description
list[float] | None

List [x, y, z] of the placed instance center, or None if failed.

Raises:

Type Description
ValueError

If instance_key already exists or room/instance not found.

Source code in embodied_gen/skills/spatial-computing/core/collector.py
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
def add_instance(
    self,
    asset_path: str,
    instance_key: str,
    in_room: str | None = None,
    on_instance: str | None = None,
    beside_instance: str | None = None,
    beside_distance: float = DEFAULT_BESIDE_DISTANCE,
    rotation_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY,
    n_max_attempt: int = DEFAULT_MAX_PLACEMENT_ATTEMPTS,
    place_strategy: Literal["top", "random"] = "random",
) -> list[float] | None:
    """Add a new instance to the scene with automatic placement.

    Args:
        asset_path: Path to the asset mesh file.
        instance_key: Unique key for the new instance.
        in_room: Optional room name to constrain placement.
        on_instance: Optional instance name to place on top of.
        beside_instance: Optional instance name to place beside (on floor).
        beside_distance: Initial buffer distance from the target instance
            for beside placement (meters). Will auto-expand if needed.
        rotation_rpy: Initial rotation in roll-pitch-yaw.
        n_max_attempt: Maximum placement attempts.
        place_strategy: Either "top" or "random".

    Returns:
        List [x, y, z] of the placed instance center, or None if failed.

    Raises:
        ValueError: If instance_key already exists or room/instance not found.

    """
    if instance_key in self.instances:
        raise ValueError(f"Instance key '{instance_key}' already exists.")

    room_poly = self._resolve_room_polygon(in_room)

    # Load mesh and compute base polygon (needed for all placement modes)
    mesh = trimesh.load(asset_path, force="mesh")
    mesh.apply_transform(
        trimesh.transformations.euler_matrix(*rotation_rpy, "sxyz")
    )

    verts = np.asarray(mesh.sample(self.mesh_sample_num))[:, :2]
    base_poly = points_to_polygon(verts)
    centroid = base_poly.centroid
    base_poly = translate(base_poly, xoff=-centroid.x, yoff=-centroid.y)

    if beside_instance is not None:
        placement = self._try_place_beside(
            base_poly=base_poly,
            beside_instance=beside_instance,
            room_poly=room_poly,
            beside_distance=beside_distance,
            n_max_attempt=n_max_attempt,
            multi_match_strategy="first",  # Default strategy
        )
        base_z = 0.0
    else:
        target_area, obstacles, base_z = self._resolve_placement_target(
            on_instance, room_poly, place_strategy
        )

        if target_area.is_empty:
            logger.error("Target area for placement is empty.")
            return None

        placement = self._try_place_polygon(
            base_poly, target_area, obstacles, n_max_attempt
        )

    if placement is None:
        logger.error(
            f"Failed to place '{instance_key}' after all attempts."
        )
        return None

    x, y, candidate = placement
    self.instances[instance_key] = candidate
    final_z = base_z - mesh.bounds[0][2] + DEFAULT_Z_OFFSET
    self._update_internal_state()

    return [round(v, 4) for v in (x, y, final_z)]
collect
collect(urdf_path: str) -> None

Parse URDF file and collect semantic information.

Parameters:

Name Type Description Default
urdf_path str

Path to the URDF file.

required
Source code in embodied_gen/skills/spatial-computing/core/collector.py
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def collect(self, urdf_path: str) -> None:
    """Parse URDF file and collect semantic information.

    Args:
        urdf_path: Path to the URDF file.

    """
    logger.info(f"Collecting URDF semantic info from {urdf_path}")
    self.urdf_path = urdf_path
    urdf_dir = os.path.dirname(urdf_path)

    self._tree = ET.parse(urdf_path)
    self._root = self._tree.getroot()

    link_transforms = self._build_link_transforms()
    self._process_links(urdf_dir, link_transforms)
    self._update_internal_state()
get_instance_center
get_instance_center(instance_key: str) -> list[float] | None

Get the center position of an instance.

Parameters:

Name Type Description Default
instance_key str

Name of the instance to query.

required

Returns:

Type Description
list[float] | None

List [x, y, z] of the instance center, or None if not found.

Source code in embodied_gen/skills/spatial-computing/core/collector.py
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
def get_instance_center(self, instance_key: str) -> list[float] | None:
    """Get the center position of an instance.

    Args:
        instance_key: Name of the instance to query.

    Returns:
        List [x, y, z] of the instance center, or None if not found.

    """
    if instance_key not in self.instances:
        logger.warning(f"Instance '{instance_key}' not found in scene.")
        return None

    # Get instance metadata
    meta = self.instance_meta.get(instance_key, {})
    xyz = meta.get("xyz", np.zeros(3))

    # Get polygon centroid for 2D position
    poly = self.instances[instance_key]
    centroid = poly.centroid

    # Return [x, y, z] where x,y are from polygon centroid, z from metadata
    center = [round(centroid.x, 4), round(centroid.y, 4), round(xyz[2], 4)]

    logger.info(f"Instance '{instance_key}' center: {center}")
    return center
remove_instance
remove_instance(instance_key: str, in_room: str | None = None) -> bool

Remove an instance from the scene.

Parameters:

Name Type Description Default
instance_key str

Exact instance name or semantic description to remove.

required
in_room str | None

Optional room constraint - only remove if instance is in this room.

None

Returns:

Type Description
bool

True if instance was removed, False if not found.

Raises:

Type Description
ValueError

If instance_key is a protected item (walls, floors).

Source code in embodied_gen/skills/spatial-computing/core/collector.py
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
def remove_instance(
    self,
    instance_key: str,
    in_room: str | None = None,
) -> bool:
    """Remove an instance from the scene.

    Args:
        instance_key: Exact instance name or semantic description to remove.
        in_room: Optional room constraint - only remove if instance is in this room.

    Returns:
        True if instance was removed, False if not found.

    Raises:
        ValueError: If instance_key is a protected item (walls, floors).

    """
    # Protect critical items
    protected = ["walls"] + [
        k for k in self.instances.keys() if "floor" in k.lower()
    ]
    if instance_key in protected:
        raise ValueError(
            f"Cannot remove protected instance '{instance_key}'. "
            f"Protected items: {protected}"
        )

    # Check if instance exists
    if instance_key not in self.instances:
        logger.warning(f"Instance '{instance_key}' not found in scene.")
        return False

    # Check room constraint if specified
    if in_room is not None:
        room_poly = self._resolve_room_polygon(in_room)
        if room_poly is not None:
            room_buffered = room_poly.buffer(0.1)
            instance_point = self.instances[
                instance_key
            ].representative_point()
            if not room_buffered.contains(instance_point):
                logger.warning(
                    f"Instance '{instance_key}' is not in room '{in_room}'."
                )
                return False

    # Remove from URDF XML tree
    if self._root is not None:
        self._remove_link_and_joint(instance_key)

    # Remove from instances dict
    del self.instances[instance_key]

    # Remove from metadata
    if instance_key in self.instance_meta:
        del self.instance_meta[instance_key]

    # Update internal state
    self._update_internal_state()

    logger.info(f"✅ Removed instance '{instance_key}' from scene.")
    return True
remove_usd_instance
remove_usd_instance(usd_path: str, output_path: str, instance_key: str) -> None

Remove an instance from a USD file.

Parameters:

Name Type Description Default
usd_path str

Path to the source USD file.

required
output_path str

Path to save the modified USD.

required
instance_key str

Prim path name of the instance to remove.

required

Raises:

Type Description
ImportError

If pxr (USD) library is not available.

Source code in embodied_gen/skills/spatial-computing/core/collector.py
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
def remove_usd_instance(
    self,
    usd_path: str,
    output_path: str,
    instance_key: str,
) -> None:
    """Remove an instance from a USD file.

    Args:
        usd_path: Path to the source USD file.
        output_path: Path to save the modified USD.
        instance_key: Prim path name of the instance to remove.

    Raises:
        ImportError: If pxr (USD) library is not available.

    """
    from pxr import Usd

    # Open USD stage
    stage = Usd.Stage.Open(usd_path)

    # Find and remove the prim. Check the defaultPrim path first, and
    # keep the old root-level path as a compatibility fallback.
    prim_paths = [self._usd_instance_prim_path(stage, instance_key)]
    legacy_prim_path = f"/{instance_key}"
    if legacy_prim_path not in prim_paths:
        prim_paths.append(legacy_prim_path)

    removed = False
    for prim_path in prim_paths:
        prim = stage.GetPrimAtPath(prim_path)
        if prim.IsValid():
            stage.RemovePrim(prim_path)
            logger.info(f"Removed prim '{prim_path}' from USD.")
            removed = True

    if not removed:
        logger.warning(
            f"Prim '{instance_key}' not found in USD stage under "
            "defaultPrim or legacy root path."
        )

    # Export modified stage
    stage.GetRootLayer().Export(output_path)
    logger.info(f"✅ Saved updated USD to {output_path}")
save_urdf
save_urdf(output_path: str) -> None

Save the current URDF tree to file.

Parameters:

Name Type Description Default
output_path str

Path to save the URDF file.

required
Source code in embodied_gen/skills/spatial-computing/core/collector.py
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
def save_urdf(self, output_path: str) -> None:
    """Save the current URDF tree to file.

    Args:
        output_path: Path to save the URDF file.

    """
    if self._tree is None:
        return

    if hasattr(ET, "indent"):
        ET.indent(self._tree, space="  ", level=0)

    self._tree.write(output_path, encoding="utf-8", xml_declaration=True)
    logger.info(f"✅ Saved updated URDF to {output_path}")
update_urdf_info
update_urdf_info(output_path: str, instance_key: str, visual_mesh_path: str, collision_mesh_path: str | None = None, trans_xyz: tuple[float, float, float] = (0, 0, 0), rot_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY, joint_type: str = 'fixed') -> None

Add a new link to the URDF tree and save.

Parameters:

Name Type Description Default
output_path str

Path to save the updated URDF.

required
instance_key str

Name for the new link.

required
visual_mesh_path str

Path to the visual mesh file.

required
collision_mesh_path str | None

Optional path to collision mesh.

None
trans_xyz tuple[float, float, float]

Translation (x, y, z).

(0, 0, 0)
rot_rpy tuple[float, float, float]

Rotation (roll, pitch, yaw).

DEFAULT_ROTATION_RPY
joint_type str

Type of joint (e.g., "fixed").

'fixed'
Source code in embodied_gen/skills/spatial-computing/core/collector.py
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
def update_urdf_info(
    self,
    output_path: str,
    instance_key: str,
    visual_mesh_path: str,
    collision_mesh_path: str | None = None,
    trans_xyz: tuple[float, float, float] = (0, 0, 0),
    rot_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY,
    joint_type: str = "fixed",
) -> None:
    """Add a new link to the URDF tree and save.

    Args:
        output_path: Path to save the updated URDF.
        instance_key: Name for the new link.
        visual_mesh_path: Path to the visual mesh file.
        collision_mesh_path: Optional path to collision mesh.
        trans_xyz: Translation (x, y, z).
        rot_rpy: Rotation (roll, pitch, yaw).
        joint_type: Type of joint (e.g., "fixed").

    """
    if self._root is None:
        return

    logger.info(f"Updating URDF for instance '{instance_key}'.")
    urdf_dir = os.path.dirname(self.urdf_path)

    # Copy mesh files
    copytree(
        os.path.dirname(visual_mesh_path),
        f"{urdf_dir}/{instance_key}",
        dirs_exist_ok=True,
    )
    visual_rel_path = (
        f"{instance_key}/{os.path.basename(visual_mesh_path)}"
    )

    collision_rel_path = None
    if collision_mesh_path is not None:
        copytree(
            os.path.dirname(collision_mesh_path),
            f"{urdf_dir}/{instance_key}",
            dirs_exist_ok=True,
        )
        collision_rel_path = (
            f"{instance_key}/{os.path.basename(collision_mesh_path)}"
        )

    # Create link element
    link = ET.SubElement(self._root, "link", attrib={"name": instance_key})

    visual = ET.SubElement(link, "visual")
    v_geo = ET.SubElement(visual, "geometry")
    ET.SubElement(v_geo, "mesh", attrib={"filename": visual_rel_path})

    if collision_rel_path is not None:
        collision = ET.SubElement(link, "collision")
        c_geo = ET.SubElement(collision, "geometry")
        ET.SubElement(
            c_geo, "mesh", attrib={"filename": collision_rel_path}
        )

    # Create joint element
    joint_name = f"joint_{instance_key}"
    joint = ET.SubElement(
        self._root,
        "joint",
        attrib={"name": joint_name, "type": joint_type},
    )

    ET.SubElement(joint, "parent", attrib={"link": "base"})
    ET.SubElement(joint, "child", attrib={"link": instance_key})

    xyz_str = f"{trans_xyz[0]:.4f} {trans_xyz[1]:.4f} {trans_xyz[2]:.4f}"
    rpy_str = f"{rot_rpy[0]:.4f} {rot_rpy[1]:.4f} {rot_rpy[2]:.4f}"
    ET.SubElement(joint, "origin", attrib={"xyz": xyz_str, "rpy": rpy_str})

    self.save_urdf(output_path)
update_usd_info
update_usd_info(usd_path: str, output_path: str, instance_key: str, visual_mesh_path: str, trans_xyz: list[float], rot_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY) -> None

Add a mesh instance to an existing USD file.

Uses Blender (bpy) to convert OBJ to USD format.

Parameters:

Name Type Description Default
usd_path str

Path to the source USD file.

required
output_path str

Path to save the modified USD.

required
instance_key str

Prim path name for the new instance.

required
visual_mesh_path str

Path to the visual mesh (OBJ format).

required
trans_xyz list[float]

Translation [x, y, z].

required
rot_rpy tuple[float, float, float]

Rotation (roll, pitch, yaw).

DEFAULT_ROTATION_RPY

Raises:

Type Description
ImportError

If pxr (USD) library or bpy is not available.

Source code in embodied_gen/skills/spatial-computing/core/collector.py
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
def update_usd_info(
    self,
    usd_path: str,
    output_path: str,
    instance_key: str,
    visual_mesh_path: str,
    trans_xyz: list[float],
    rot_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY,
) -> None:
    """Add a mesh instance to an existing USD file.

    Uses Blender (bpy) to convert OBJ to USD format.

    Args:
        usd_path: Path to the source USD file.
        output_path: Path to save the modified USD.
        instance_key: Prim path name for the new instance.
        visual_mesh_path: Path to the visual mesh (OBJ format).
        trans_xyz: Translation [x, y, z].
        rot_rpy: Rotation (roll, pitch, yaw).

    Raises:
        ImportError: If pxr (USD) library or bpy is not available.

    """
    import bpy
    from pxr import Gf, Usd, UsdGeom

    out_dir = os.path.dirname(output_path)
    target_dir = os.path.join(out_dir, instance_key)
    os.makedirs(target_dir, exist_ok=True)

    mesh_filename = os.path.basename(visual_mesh_path)
    usdc_filename = os.path.splitext(mesh_filename)[0] + ".usdc"
    target_usdc_path = os.path.join(target_dir, usdc_filename)

    logger.info(
        f"Converting with Blender (bpy): "
        f"{visual_mesh_path} -> {target_usdc_path}"
    )
    bpy.ops.wm.read_factory_settings(use_empty=True)
    bpy.ops.wm.obj_import(
        filepath=visual_mesh_path,
        forward_axis="Y",
        up_axis="Z",
    )
    bpy.ops.wm.usd_export(
        filepath=target_usdc_path,
        selected_objects_only=False,
    )

    # Copy texture files
    src_dir = os.path.dirname(visual_mesh_path)
    for f in os.listdir(src_dir):
        if f.lower().endswith((".png", ".jpg", ".jpeg", ".mtl")):
            copy2(os.path.join(src_dir, f), target_dir)

    final_rel_path = f"./{instance_key}/{usdc_filename}"

    # Update USD stage
    stage = Usd.Stage.Open(usd_path)
    prim_path = self._usd_instance_prim_path(stage, instance_key)
    mesh_prim = UsdGeom.Xform.Define(stage, prim_path)

    ref_prim = UsdGeom.Mesh.Define(stage, f"{prim_path}/Mesh")
    ref_prim.GetPrim().GetReferences().AddReference(final_rel_path)

    # Build transform matrix
    translation_mat = Gf.Matrix4d().SetTranslate(
        Gf.Vec3d(trans_xyz[0], trans_xyz[1], trans_xyz[2])
    )
    rx = Gf.Matrix4d().SetRotate(
        Gf.Rotation(Gf.Vec3d(1, 0, 0), np.degrees(rot_rpy[0]))
    )
    ry = Gf.Matrix4d().SetRotate(
        Gf.Rotation(Gf.Vec3d(0, 1, 0), np.degrees(rot_rpy[1]))
    )
    rz = Gf.Matrix4d().SetRotate(
        Gf.Rotation(Gf.Vec3d(0, 0, 1), np.degrees(rot_rpy[2]))
    )
    rotation_mat = rx * ry * rz
    transform = rotation_mat * translation_mat
    mesh_prim.AddTransformOp().Set(transform)

    stage.GetRootLayer().Export(output_path)
    logger.info(f"✅ Saved updated USD to {output_path}")

get_actionable_surface

get_actionable_surface(mesh: Trimesh, tol_angle: int = 10, tol_z: float = 0.02, area_tolerance: float = 0.15, place_strategy: Literal['top', 'random'] = 'random') -> tuple[float, Geometry]

Extract the actionable (placeable) surface from a mesh.

Finds upward-facing surfaces and returns the best one based on the placement strategy.

Parameters:

Name Type Description Default
mesh Trimesh

The input trimesh object.

required
tol_angle int

Angle tolerance in degrees for detecting up-facing normals.

10
tol_z float

Z-coordinate tolerance for clustering faces.

0.02
area_tolerance float

Tolerance for selecting candidate surfaces by area.

0.15
place_strategy Literal['top', 'random']

Either "top" (highest surface) or "random".

'random'

Returns:

Type Description
float

A tuple of (z_height, surface_polygon) representing the selected

Geometry

actionable surface.

Source code in embodied_gen/skills/spatial-computing/core/geometry.py
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def get_actionable_surface(
    mesh: trimesh.Trimesh,
    tol_angle: int = 10,
    tol_z: float = 0.02,
    area_tolerance: float = 0.15,
    place_strategy: Literal["top", "random"] = "random",
) -> tuple[float, Geometry]:
    """Extract the actionable (placeable) surface from a mesh.

    Finds upward-facing surfaces and returns the best one based on the
    placement strategy.

    Args:
        mesh: The input trimesh object.
        tol_angle: Angle tolerance in degrees for detecting up-facing normals.
        tol_z: Z-coordinate tolerance for clustering faces.
        area_tolerance: Tolerance for selecting candidate surfaces by area.
        place_strategy: Either "top" (highest surface) or "random".

    Returns:
        A tuple of (z_height, surface_polygon) representing the selected
        actionable surface.

    """
    up_vec = np.array([0, 0, 1])
    dots = np.dot(mesh.face_normals, up_vec)
    valid_mask = dots > np.cos(np.deg2rad(tol_angle))

    if not np.any(valid_mask):
        logger.warning(
            "No up-facing surfaces found. Falling back to bounding box top."
        )
        verts = mesh.vertices[:, :2]
        return mesh.bounds[1][2], MultiPoint(verts).convex_hull

    valid_faces_indices = np.where(valid_mask)[0]
    face_z = mesh.triangles_center[valid_mask][:, 2]
    face_areas = mesh.area_faces[valid_mask]

    z_clusters = _cluster_faces_by_z(
        face_z, face_areas, valid_faces_indices, tol_z
    )

    if not z_clusters:
        return mesh.bounds[1][2], MultiPoint(mesh.vertices[:, :2]).convex_hull

    selected_z, selected_data = _select_surface_cluster(
        z_clusters, area_tolerance, place_strategy
    )

    # For "top" strategy, use the highest z among all clusters for
    # base height, while keeping the largest-area polygon for XY placement.
    if place_strategy == "top":
        highest_z = max(z_clusters.keys())
        if highest_z > selected_z:
            logger.info(
                f"Overriding base Z from {selected_z:.3f} to "
                f"highest surface {highest_z:.3f}"
            )
            selected_z = highest_z

    cluster_faces = mesh.faces[selected_data["indices"]]
    temp_mesh = trimesh.Trimesh(vertices=mesh.vertices, faces=cluster_faces)
    samples, _ = trimesh.sample.sample_surface(temp_mesh, 10000)

    if len(samples) < 3:
        logger.warning(
            f"Failed to sample enough points on layer Z={selected_z}. "
            "Returning empty polygon."
        )
        return selected_z, Polygon()

    surface_poly = MultiPoint(samples[:, :2]).convex_hull
    return selected_z, surface_poly

points_to_polygon

points_to_polygon(points: ndarray, smooth_thresh: float = 0.2, scanline_step: float = 0.01) -> Polygon

Convert point clouds into polygon contours using sweep line algorithm.

Parameters:

Name Type Description Default
points ndarray

Array of 2D points with shape (N, 2).

required
smooth_thresh float

Buffer threshold for smoothing the polygon.

0.2
scanline_step float

Step size for the scanline sweep.

0.01

Returns:

Type Description
Polygon

A Shapely Polygon representing the contour of the point cloud.

Source code in embodied_gen/skills/spatial-computing/core/geometry.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def points_to_polygon(
    points: np.ndarray,
    smooth_thresh: float = 0.2,
    scanline_step: float = 0.01,
) -> Polygon:
    """Convert point clouds into polygon contours using sweep line algorithm.

    Args:
        points: Array of 2D points with shape (N, 2).
        smooth_thresh: Buffer threshold for smoothing the polygon.
        scanline_step: Step size for the scanline sweep.

    Returns:
        A Shapely Polygon representing the contour of the point cloud.

    """
    if len(points) == 0:
        return Polygon()

    ys = points[:, 1]
    y_min, y_max = ys.min(), ys.max()
    y_values = np.arange(y_min, y_max + scanline_step, scanline_step)

    upper: list[list[float]] = []
    lower: list[list[float]] = []

    for y in y_values:
        pts_in_strip = points[(ys >= y) & (ys < y + scanline_step)]
        if len(pts_in_strip) == 0:
            continue

        xs = pts_in_strip[:, 0]
        upper.append([xs.max(), y])
        lower.append([xs.min(), y])

    contour = upper + lower[::-1]
    if len(contour) < 3:
        return Polygon()

    poly = Polygon(contour)
    return poly.buffer(smooth_thresh).buffer(-smooth_thresh)

embodied_gen.skills.spatial-computing.core.geometry

get_actionable_surface

get_actionable_surface(mesh: Trimesh, tol_angle: int = 10, tol_z: float = 0.02, area_tolerance: float = 0.15, place_strategy: Literal['top', 'random'] = 'random') -> tuple[float, Geometry]

Extract the actionable (placeable) surface from a mesh.

Finds upward-facing surfaces and returns the best one based on the placement strategy.

Parameters:

Name Type Description Default
mesh Trimesh

The input trimesh object.

required
tol_angle int

Angle tolerance in degrees for detecting up-facing normals.

10
tol_z float

Z-coordinate tolerance for clustering faces.

0.02
area_tolerance float

Tolerance for selecting candidate surfaces by area.

0.15
place_strategy Literal['top', 'random']

Either "top" (highest surface) or "random".

'random'

Returns:

Type Description
float

A tuple of (z_height, surface_polygon) representing the selected

Geometry

actionable surface.

Source code in embodied_gen/skills/spatial-computing/core/geometry.py
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def get_actionable_surface(
    mesh: trimesh.Trimesh,
    tol_angle: int = 10,
    tol_z: float = 0.02,
    area_tolerance: float = 0.15,
    place_strategy: Literal["top", "random"] = "random",
) -> tuple[float, Geometry]:
    """Extract the actionable (placeable) surface from a mesh.

    Finds upward-facing surfaces and returns the best one based on the
    placement strategy.

    Args:
        mesh: The input trimesh object.
        tol_angle: Angle tolerance in degrees for detecting up-facing normals.
        tol_z: Z-coordinate tolerance for clustering faces.
        area_tolerance: Tolerance for selecting candidate surfaces by area.
        place_strategy: Either "top" (highest surface) or "random".

    Returns:
        A tuple of (z_height, surface_polygon) representing the selected
        actionable surface.

    """
    up_vec = np.array([0, 0, 1])
    dots = np.dot(mesh.face_normals, up_vec)
    valid_mask = dots > np.cos(np.deg2rad(tol_angle))

    if not np.any(valid_mask):
        logger.warning(
            "No up-facing surfaces found. Falling back to bounding box top."
        )
        verts = mesh.vertices[:, :2]
        return mesh.bounds[1][2], MultiPoint(verts).convex_hull

    valid_faces_indices = np.where(valid_mask)[0]
    face_z = mesh.triangles_center[valid_mask][:, 2]
    face_areas = mesh.area_faces[valid_mask]

    z_clusters = _cluster_faces_by_z(
        face_z, face_areas, valid_faces_indices, tol_z
    )

    if not z_clusters:
        return mesh.bounds[1][2], MultiPoint(mesh.vertices[:, :2]).convex_hull

    selected_z, selected_data = _select_surface_cluster(
        z_clusters, area_tolerance, place_strategy
    )

    # For "top" strategy, use the highest z among all clusters for
    # base height, while keeping the largest-area polygon for XY placement.
    if place_strategy == "top":
        highest_z = max(z_clusters.keys())
        if highest_z > selected_z:
            logger.info(
                f"Overriding base Z from {selected_z:.3f} to "
                f"highest surface {highest_z:.3f}"
            )
            selected_z = highest_z

    cluster_faces = mesh.faces[selected_data["indices"]]
    temp_mesh = trimesh.Trimesh(vertices=mesh.vertices, faces=cluster_faces)
    samples, _ = trimesh.sample.sample_surface(temp_mesh, 10000)

    if len(samples) < 3:
        logger.warning(
            f"Failed to sample enough points on layer Z={selected_z}. "
            "Returning empty polygon."
        )
        return selected_z, Polygon()

    surface_poly = MultiPoint(samples[:, :2]).convex_hull
    return selected_z, surface_poly

points_to_polygon

points_to_polygon(points: ndarray, smooth_thresh: float = 0.2, scanline_step: float = 0.01) -> Polygon

Convert point clouds into polygon contours using sweep line algorithm.

Parameters:

Name Type Description Default
points ndarray

Array of 2D points with shape (N, 2).

required
smooth_thresh float

Buffer threshold for smoothing the polygon.

0.2
scanline_step float

Step size for the scanline sweep.

0.01

Returns:

Type Description
Polygon

A Shapely Polygon representing the contour of the point cloud.

Source code in embodied_gen/skills/spatial-computing/core/geometry.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def points_to_polygon(
    points: np.ndarray,
    smooth_thresh: float = 0.2,
    scanline_step: float = 0.01,
) -> Polygon:
    """Convert point clouds into polygon contours using sweep line algorithm.

    Args:
        points: Array of 2D points with shape (N, 2).
        smooth_thresh: Buffer threshold for smoothing the polygon.
        scanline_step: Step size for the scanline sweep.

    Returns:
        A Shapely Polygon representing the contour of the point cloud.

    """
    if len(points) == 0:
        return Polygon()

    ys = points[:, 1]
    y_min, y_max = ys.min(), ys.max()
    y_values = np.arange(y_min, y_max + scanline_step, scanline_step)

    upper: list[list[float]] = []
    lower: list[list[float]] = []

    for y in y_values:
        pts_in_strip = points[(ys >= y) & (ys < y + scanline_step)]
        if len(pts_in_strip) == 0:
            continue

        xs = pts_in_strip[:, 0]
        upper.append([xs.max(), y])
        lower.append([xs.min(), y])

    contour = upper + lower[::-1]
    if len(contour) < 3:
        return Polygon()

    poly = Polygon(contour)
    return poly.buffer(smooth_thresh).buffer(-smooth_thresh)

embodied_gen.skills.spatial-computing.core.collector

UrdfSemanticInfoCollector

UrdfSemanticInfoCollector(mesh_sample_num: int = DEFAULT_MESH_SAMPLE_NUM, ignore_items: list[str] | None = None)

Collector for URDF semantic information.

Parses URDF files to extract room layouts, object footprints, and provides methods for adding new instances and updating URDF/USD files.

Attributes:

Name Type Description
mesh_sample_num

Number of points to sample from meshes.

ignore_items

List of item name patterns to ignore.

instances dict[str, Polygon]

Dictionary of instance name to footprint polygon.

instance_meta dict[str, dict]

Dictionary of instance metadata (mesh path, pose).

rooms dict[str, Geometry]

Dictionary of room polygons.

footprints dict[str, Geometry]

Dictionary of object footprints.

occ_area Geometry

Union of all occupied areas.

floor_union Geometry

Union of all floor polygons.

Initialize the collector.

Parameters:

Name Type Description Default
mesh_sample_num int

Number of points to sample from meshes.

DEFAULT_MESH_SAMPLE_NUM
ignore_items list[str] | None

List of item name patterns to ignore during parsing.

None
Source code in embodied_gen/skills/spatial-computing/core/collector.py
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def __init__(
    self,
    mesh_sample_num: int = DEFAULT_MESH_SAMPLE_NUM,
    ignore_items: list[str] | None = None,
) -> None:
    """Initialize the collector.

    Args:
        mesh_sample_num: Number of points to sample from meshes.
        ignore_items: List of item name patterns to ignore during parsing.

    """
    self.mesh_sample_num = mesh_sample_num
    self.ignore_items = ignore_items or list(DEFAULT_IGNORE_ITEMS)

    self.instances: dict[str, Polygon] = {}
    self.instance_meta: dict[str, dict] = {}
    self.rooms: dict[str, Geometry] = {}
    self.footprints: dict[str, Geometry] = {}
    self.occ_area: Geometry = Polygon()
    self.floor_union: Geometry = Polygon()

    self.urdf_path: str = ""
    self._tree: ET.ElementTree | None = None
    self._root: ET.Element | None = None
add_instance
add_instance(asset_path: str, instance_key: str, in_room: str | None = None, on_instance: str | None = None, beside_instance: str | None = None, beside_distance: float = DEFAULT_BESIDE_DISTANCE, rotation_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY, n_max_attempt: int = DEFAULT_MAX_PLACEMENT_ATTEMPTS, place_strategy: Literal['top', 'random'] = 'random') -> list[float] | None

Add a new instance to the scene with automatic placement.

Parameters:

Name Type Description Default
asset_path str

Path to the asset mesh file.

required
instance_key str

Unique key for the new instance.

required
in_room str | None

Optional room name to constrain placement.

None
on_instance str | None

Optional instance name to place on top of.

None
beside_instance str | None

Optional instance name to place beside (on floor).

None
beside_distance float

Initial buffer distance from the target instance for beside placement (meters). Will auto-expand if needed.

DEFAULT_BESIDE_DISTANCE
rotation_rpy tuple[float, float, float]

Initial rotation in roll-pitch-yaw.

DEFAULT_ROTATION_RPY
n_max_attempt int

Maximum placement attempts.

DEFAULT_MAX_PLACEMENT_ATTEMPTS
place_strategy Literal['top', 'random']

Either "top" or "random".

'random'

Returns:

Type Description
list[float] | None

List [x, y, z] of the placed instance center, or None if failed.

Raises:

Type Description
ValueError

If instance_key already exists or room/instance not found.

Source code in embodied_gen/skills/spatial-computing/core/collector.py
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
def add_instance(
    self,
    asset_path: str,
    instance_key: str,
    in_room: str | None = None,
    on_instance: str | None = None,
    beside_instance: str | None = None,
    beside_distance: float = DEFAULT_BESIDE_DISTANCE,
    rotation_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY,
    n_max_attempt: int = DEFAULT_MAX_PLACEMENT_ATTEMPTS,
    place_strategy: Literal["top", "random"] = "random",
) -> list[float] | None:
    """Add a new instance to the scene with automatic placement.

    Args:
        asset_path: Path to the asset mesh file.
        instance_key: Unique key for the new instance.
        in_room: Optional room name to constrain placement.
        on_instance: Optional instance name to place on top of.
        beside_instance: Optional instance name to place beside (on floor).
        beside_distance: Initial buffer distance from the target instance
            for beside placement (meters). Will auto-expand if needed.
        rotation_rpy: Initial rotation in roll-pitch-yaw.
        n_max_attempt: Maximum placement attempts.
        place_strategy: Either "top" or "random".

    Returns:
        List [x, y, z] of the placed instance center, or None if failed.

    Raises:
        ValueError: If instance_key already exists or room/instance not found.

    """
    if instance_key in self.instances:
        raise ValueError(f"Instance key '{instance_key}' already exists.")

    room_poly = self._resolve_room_polygon(in_room)

    # Load mesh and compute base polygon (needed for all placement modes)
    mesh = trimesh.load(asset_path, force="mesh")
    mesh.apply_transform(
        trimesh.transformations.euler_matrix(*rotation_rpy, "sxyz")
    )

    verts = np.asarray(mesh.sample(self.mesh_sample_num))[:, :2]
    base_poly = points_to_polygon(verts)
    centroid = base_poly.centroid
    base_poly = translate(base_poly, xoff=-centroid.x, yoff=-centroid.y)

    if beside_instance is not None:
        placement = self._try_place_beside(
            base_poly=base_poly,
            beside_instance=beside_instance,
            room_poly=room_poly,
            beside_distance=beside_distance,
            n_max_attempt=n_max_attempt,
            multi_match_strategy="first",  # Default strategy
        )
        base_z = 0.0
    else:
        target_area, obstacles, base_z = self._resolve_placement_target(
            on_instance, room_poly, place_strategy
        )

        if target_area.is_empty:
            logger.error("Target area for placement is empty.")
            return None

        placement = self._try_place_polygon(
            base_poly, target_area, obstacles, n_max_attempt
        )

    if placement is None:
        logger.error(
            f"Failed to place '{instance_key}' after all attempts."
        )
        return None

    x, y, candidate = placement
    self.instances[instance_key] = candidate
    final_z = base_z - mesh.bounds[0][2] + DEFAULT_Z_OFFSET
    self._update_internal_state()

    return [round(v, 4) for v in (x, y, final_z)]
collect
collect(urdf_path: str) -> None

Parse URDF file and collect semantic information.

Parameters:

Name Type Description Default
urdf_path str

Path to the URDF file.

required
Source code in embodied_gen/skills/spatial-computing/core/collector.py
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def collect(self, urdf_path: str) -> None:
    """Parse URDF file and collect semantic information.

    Args:
        urdf_path: Path to the URDF file.

    """
    logger.info(f"Collecting URDF semantic info from {urdf_path}")
    self.urdf_path = urdf_path
    urdf_dir = os.path.dirname(urdf_path)

    self._tree = ET.parse(urdf_path)
    self._root = self._tree.getroot()

    link_transforms = self._build_link_transforms()
    self._process_links(urdf_dir, link_transforms)
    self._update_internal_state()
get_instance_center
get_instance_center(instance_key: str) -> list[float] | None

Get the center position of an instance.

Parameters:

Name Type Description Default
instance_key str

Name of the instance to query.

required

Returns:

Type Description
list[float] | None

List [x, y, z] of the instance center, or None if not found.

Source code in embodied_gen/skills/spatial-computing/core/collector.py
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
def get_instance_center(self, instance_key: str) -> list[float] | None:
    """Get the center position of an instance.

    Args:
        instance_key: Name of the instance to query.

    Returns:
        List [x, y, z] of the instance center, or None if not found.

    """
    if instance_key not in self.instances:
        logger.warning(f"Instance '{instance_key}' not found in scene.")
        return None

    # Get instance metadata
    meta = self.instance_meta.get(instance_key, {})
    xyz = meta.get("xyz", np.zeros(3))

    # Get polygon centroid for 2D position
    poly = self.instances[instance_key]
    centroid = poly.centroid

    # Return [x, y, z] where x,y are from polygon centroid, z from metadata
    center = [round(centroid.x, 4), round(centroid.y, 4), round(xyz[2], 4)]

    logger.info(f"Instance '{instance_key}' center: {center}")
    return center
remove_instance
remove_instance(instance_key: str, in_room: str | None = None) -> bool

Remove an instance from the scene.

Parameters:

Name Type Description Default
instance_key str

Exact instance name or semantic description to remove.

required
in_room str | None

Optional room constraint - only remove if instance is in this room.

None

Returns:

Type Description
bool

True if instance was removed, False if not found.

Raises:

Type Description
ValueError

If instance_key is a protected item (walls, floors).

Source code in embodied_gen/skills/spatial-computing/core/collector.py
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
def remove_instance(
    self,
    instance_key: str,
    in_room: str | None = None,
) -> bool:
    """Remove an instance from the scene.

    Args:
        instance_key: Exact instance name or semantic description to remove.
        in_room: Optional room constraint - only remove if instance is in this room.

    Returns:
        True if instance was removed, False if not found.

    Raises:
        ValueError: If instance_key is a protected item (walls, floors).

    """
    # Protect critical items
    protected = ["walls"] + [
        k for k in self.instances.keys() if "floor" in k.lower()
    ]
    if instance_key in protected:
        raise ValueError(
            f"Cannot remove protected instance '{instance_key}'. "
            f"Protected items: {protected}"
        )

    # Check if instance exists
    if instance_key not in self.instances:
        logger.warning(f"Instance '{instance_key}' not found in scene.")
        return False

    # Check room constraint if specified
    if in_room is not None:
        room_poly = self._resolve_room_polygon(in_room)
        if room_poly is not None:
            room_buffered = room_poly.buffer(0.1)
            instance_point = self.instances[
                instance_key
            ].representative_point()
            if not room_buffered.contains(instance_point):
                logger.warning(
                    f"Instance '{instance_key}' is not in room '{in_room}'."
                )
                return False

    # Remove from URDF XML tree
    if self._root is not None:
        self._remove_link_and_joint(instance_key)

    # Remove from instances dict
    del self.instances[instance_key]

    # Remove from metadata
    if instance_key in self.instance_meta:
        del self.instance_meta[instance_key]

    # Update internal state
    self._update_internal_state()

    logger.info(f"✅ Removed instance '{instance_key}' from scene.")
    return True
remove_usd_instance
remove_usd_instance(usd_path: str, output_path: str, instance_key: str) -> None

Remove an instance from a USD file.

Parameters:

Name Type Description Default
usd_path str

Path to the source USD file.

required
output_path str

Path to save the modified USD.

required
instance_key str

Prim path name of the instance to remove.

required

Raises:

Type Description
ImportError

If pxr (USD) library is not available.

Source code in embodied_gen/skills/spatial-computing/core/collector.py
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
def remove_usd_instance(
    self,
    usd_path: str,
    output_path: str,
    instance_key: str,
) -> None:
    """Remove an instance from a USD file.

    Args:
        usd_path: Path to the source USD file.
        output_path: Path to save the modified USD.
        instance_key: Prim path name of the instance to remove.

    Raises:
        ImportError: If pxr (USD) library is not available.

    """
    from pxr import Usd

    # Open USD stage
    stage = Usd.Stage.Open(usd_path)

    # Find and remove the prim. Check the defaultPrim path first, and
    # keep the old root-level path as a compatibility fallback.
    prim_paths = [self._usd_instance_prim_path(stage, instance_key)]
    legacy_prim_path = f"/{instance_key}"
    if legacy_prim_path not in prim_paths:
        prim_paths.append(legacy_prim_path)

    removed = False
    for prim_path in prim_paths:
        prim = stage.GetPrimAtPath(prim_path)
        if prim.IsValid():
            stage.RemovePrim(prim_path)
            logger.info(f"Removed prim '{prim_path}' from USD.")
            removed = True

    if not removed:
        logger.warning(
            f"Prim '{instance_key}' not found in USD stage under "
            "defaultPrim or legacy root path."
        )

    # Export modified stage
    stage.GetRootLayer().Export(output_path)
    logger.info(f"✅ Saved updated USD to {output_path}")
save_urdf
save_urdf(output_path: str) -> None

Save the current URDF tree to file.

Parameters:

Name Type Description Default
output_path str

Path to save the URDF file.

required
Source code in embodied_gen/skills/spatial-computing/core/collector.py
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
def save_urdf(self, output_path: str) -> None:
    """Save the current URDF tree to file.

    Args:
        output_path: Path to save the URDF file.

    """
    if self._tree is None:
        return

    if hasattr(ET, "indent"):
        ET.indent(self._tree, space="  ", level=0)

    self._tree.write(output_path, encoding="utf-8", xml_declaration=True)
    logger.info(f"✅ Saved updated URDF to {output_path}")
update_urdf_info
update_urdf_info(output_path: str, instance_key: str, visual_mesh_path: str, collision_mesh_path: str | None = None, trans_xyz: tuple[float, float, float] = (0, 0, 0), rot_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY, joint_type: str = 'fixed') -> None

Add a new link to the URDF tree and save.

Parameters:

Name Type Description Default
output_path str

Path to save the updated URDF.

required
instance_key str

Name for the new link.

required
visual_mesh_path str

Path to the visual mesh file.

required
collision_mesh_path str | None

Optional path to collision mesh.

None
trans_xyz tuple[float, float, float]

Translation (x, y, z).

(0, 0, 0)
rot_rpy tuple[float, float, float]

Rotation (roll, pitch, yaw).

DEFAULT_ROTATION_RPY
joint_type str

Type of joint (e.g., "fixed").

'fixed'
Source code in embodied_gen/skills/spatial-computing/core/collector.py
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
def update_urdf_info(
    self,
    output_path: str,
    instance_key: str,
    visual_mesh_path: str,
    collision_mesh_path: str | None = None,
    trans_xyz: tuple[float, float, float] = (0, 0, 0),
    rot_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY,
    joint_type: str = "fixed",
) -> None:
    """Add a new link to the URDF tree and save.

    Args:
        output_path: Path to save the updated URDF.
        instance_key: Name for the new link.
        visual_mesh_path: Path to the visual mesh file.
        collision_mesh_path: Optional path to collision mesh.
        trans_xyz: Translation (x, y, z).
        rot_rpy: Rotation (roll, pitch, yaw).
        joint_type: Type of joint (e.g., "fixed").

    """
    if self._root is None:
        return

    logger.info(f"Updating URDF for instance '{instance_key}'.")
    urdf_dir = os.path.dirname(self.urdf_path)

    # Copy mesh files
    copytree(
        os.path.dirname(visual_mesh_path),
        f"{urdf_dir}/{instance_key}",
        dirs_exist_ok=True,
    )
    visual_rel_path = (
        f"{instance_key}/{os.path.basename(visual_mesh_path)}"
    )

    collision_rel_path = None
    if collision_mesh_path is not None:
        copytree(
            os.path.dirname(collision_mesh_path),
            f"{urdf_dir}/{instance_key}",
            dirs_exist_ok=True,
        )
        collision_rel_path = (
            f"{instance_key}/{os.path.basename(collision_mesh_path)}"
        )

    # Create link element
    link = ET.SubElement(self._root, "link", attrib={"name": instance_key})

    visual = ET.SubElement(link, "visual")
    v_geo = ET.SubElement(visual, "geometry")
    ET.SubElement(v_geo, "mesh", attrib={"filename": visual_rel_path})

    if collision_rel_path is not None:
        collision = ET.SubElement(link, "collision")
        c_geo = ET.SubElement(collision, "geometry")
        ET.SubElement(
            c_geo, "mesh", attrib={"filename": collision_rel_path}
        )

    # Create joint element
    joint_name = f"joint_{instance_key}"
    joint = ET.SubElement(
        self._root,
        "joint",
        attrib={"name": joint_name, "type": joint_type},
    )

    ET.SubElement(joint, "parent", attrib={"link": "base"})
    ET.SubElement(joint, "child", attrib={"link": instance_key})

    xyz_str = f"{trans_xyz[0]:.4f} {trans_xyz[1]:.4f} {trans_xyz[2]:.4f}"
    rpy_str = f"{rot_rpy[0]:.4f} {rot_rpy[1]:.4f} {rot_rpy[2]:.4f}"
    ET.SubElement(joint, "origin", attrib={"xyz": xyz_str, "rpy": rpy_str})

    self.save_urdf(output_path)
update_usd_info
update_usd_info(usd_path: str, output_path: str, instance_key: str, visual_mesh_path: str, trans_xyz: list[float], rot_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY) -> None

Add a mesh instance to an existing USD file.

Uses Blender (bpy) to convert OBJ to USD format.

Parameters:

Name Type Description Default
usd_path str

Path to the source USD file.

required
output_path str

Path to save the modified USD.

required
instance_key str

Prim path name for the new instance.

required
visual_mesh_path str

Path to the visual mesh (OBJ format).

required
trans_xyz list[float]

Translation [x, y, z].

required
rot_rpy tuple[float, float, float]

Rotation (roll, pitch, yaw).

DEFAULT_ROTATION_RPY

Raises:

Type Description
ImportError

If pxr (USD) library or bpy is not available.

Source code in embodied_gen/skills/spatial-computing/core/collector.py
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
def update_usd_info(
    self,
    usd_path: str,
    output_path: str,
    instance_key: str,
    visual_mesh_path: str,
    trans_xyz: list[float],
    rot_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY,
) -> None:
    """Add a mesh instance to an existing USD file.

    Uses Blender (bpy) to convert OBJ to USD format.

    Args:
        usd_path: Path to the source USD file.
        output_path: Path to save the modified USD.
        instance_key: Prim path name for the new instance.
        visual_mesh_path: Path to the visual mesh (OBJ format).
        trans_xyz: Translation [x, y, z].
        rot_rpy: Rotation (roll, pitch, yaw).

    Raises:
        ImportError: If pxr (USD) library or bpy is not available.

    """
    import bpy
    from pxr import Gf, Usd, UsdGeom

    out_dir = os.path.dirname(output_path)
    target_dir = os.path.join(out_dir, instance_key)
    os.makedirs(target_dir, exist_ok=True)

    mesh_filename = os.path.basename(visual_mesh_path)
    usdc_filename = os.path.splitext(mesh_filename)[0] + ".usdc"
    target_usdc_path = os.path.join(target_dir, usdc_filename)

    logger.info(
        f"Converting with Blender (bpy): "
        f"{visual_mesh_path} -> {target_usdc_path}"
    )
    bpy.ops.wm.read_factory_settings(use_empty=True)
    bpy.ops.wm.obj_import(
        filepath=visual_mesh_path,
        forward_axis="Y",
        up_axis="Z",
    )
    bpy.ops.wm.usd_export(
        filepath=target_usdc_path,
        selected_objects_only=False,
    )

    # Copy texture files
    src_dir = os.path.dirname(visual_mesh_path)
    for f in os.listdir(src_dir):
        if f.lower().endswith((".png", ".jpg", ".jpeg", ".mtl")):
            copy2(os.path.join(src_dir, f), target_dir)

    final_rel_path = f"./{instance_key}/{usdc_filename}"

    # Update USD stage
    stage = Usd.Stage.Open(usd_path)
    prim_path = self._usd_instance_prim_path(stage, instance_key)
    mesh_prim = UsdGeom.Xform.Define(stage, prim_path)

    ref_prim = UsdGeom.Mesh.Define(stage, f"{prim_path}/Mesh")
    ref_prim.GetPrim().GetReferences().AddReference(final_rel_path)

    # Build transform matrix
    translation_mat = Gf.Matrix4d().SetTranslate(
        Gf.Vec3d(trans_xyz[0], trans_xyz[1], trans_xyz[2])
    )
    rx = Gf.Matrix4d().SetRotate(
        Gf.Rotation(Gf.Vec3d(1, 0, 0), np.degrees(rot_rpy[0]))
    )
    ry = Gf.Matrix4d().SetRotate(
        Gf.Rotation(Gf.Vec3d(0, 1, 0), np.degrees(rot_rpy[1]))
    )
    rz = Gf.Matrix4d().SetRotate(
        Gf.Rotation(Gf.Vec3d(0, 0, 1), np.degrees(rot_rpy[2]))
    )
    rotation_mat = rx * ry * rz
    transform = rotation_mat * translation_mat
    mesh_prim.AddTransformOp().Set(transform)

    stage.GetRootLayer().Export(output_path)
    logger.info(f"✅ Saved updated USD to {output_path}")

embodied_gen.skills.spatial-computing.core.visualizer

FloorplanVisualizer

Static utility class for visualizing floorplans.

draw_poly staticmethod
draw_poly(ax: Axes, poly: Geometry, **kwargs) -> None

Draw a polygon or multi-polygon on matplotlib axes.

Parameters:

Name Type Description Default
ax Axes

Matplotlib axes object.

required
poly Geometry

Shapely Polygon or MultiPolygon to draw.

required
**kwargs

Additional arguments passed to ax.fill().

{}
Source code in embodied_gen/skills/spatial-computing/core/visualizer.py
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@staticmethod
def draw_poly(ax: Axes, poly: Geometry, **kwargs) -> None:
    """Draw a polygon or multi-polygon on matplotlib axes.

    Args:
        ax: Matplotlib axes object.
        poly: Shapely Polygon or MultiPolygon to draw.
        **kwargs: Additional arguments passed to ax.fill().

    """
    if poly.is_empty:
        return

    geoms = poly.geoms if hasattr(poly, "geoms") else [poly]

    color = kwargs.pop("color", None)
    if color is None:
        cmap = plt.get_cmap("tab10")
        colors = [cmap(i) for i in range(len(geoms))]
    else:
        colors = [color] * len(geoms)

    for i, p in enumerate(geoms):
        if p.is_empty:
            continue
        x, y = p.exterior.xy
        ax.fill(x, y, facecolor=colors[i], **kwargs)
plot classmethod
plot(rooms: dict[str, Geometry], footprints: dict[str, Geometry], occ_area: Geometry, save_path: str, trajectory: ndarray | None = None, arrow_stride: int = 10, current_index: int | None = None, point_markers: bool = True, dpi: int = 300) -> None

Generate and save a floorplan visualization.

Parameters:

Name Type Description Default
rooms dict[str, Geometry]

Dictionary mapping room names to floor polygons.

required
footprints dict[str, Geometry]

Dictionary mapping object names to footprint polygons.

required
occ_area Geometry

Union of all occupied areas.

required
save_path str

Path to save the output image.

required
trajectory ndarray | None

Optional (N, 2) or (N, 3) array of waypoints. When the third column (rot_deg, tangent heading) is present, heading arrows are drawn. Rendered as a red curve overlay.

None
arrow_stride int

Draw a heading arrow every arrow_stride points (0 disables arrows). Ignored when current_index is set.

10
current_index int | None

Animation frame index. When set, only the traveled path (up to this index) is drawn, with a green dot at the current position and a red heading arrow; the future path is hidden.

None
point_markers bool

When True, mark every trajectory point with a small red dot (in addition to the curve).

True
dpi int

Output image resolution in dots per inch.

300
Source code in embodied_gen/skills/spatial-computing/core/visualizer.py
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@classmethod
def plot(
    cls,
    rooms: dict[str, Geometry],
    footprints: dict[str, Geometry],
    occ_area: Geometry,
    save_path: str,
    trajectory: np.ndarray | None = None,
    arrow_stride: int = 10,
    current_index: int | None = None,
    point_markers: bool = True,
    dpi: int = 300,
) -> None:
    """Generate and save a floorplan visualization.

    Args:
        rooms: Dictionary mapping room names to floor polygons.
        footprints: Dictionary mapping object names to footprint polygons.
        occ_area: Union of all occupied areas.
        save_path: Path to save the output image.
        trajectory: Optional (N, 2) or (N, 3) array of waypoints. When the
            third column (rot_deg, tangent heading) is present, heading
            arrows are drawn. Rendered as a red curve overlay.
        arrow_stride: Draw a heading arrow every ``arrow_stride`` points
            (0 disables arrows). Ignored when ``current_index`` is set.
        current_index: Animation frame index. When set, only the traveled
            path (up to this index) is drawn, with a green dot at the
            current position and a red heading arrow; the future path is
            hidden.
        point_markers: When True, mark every trajectory point with a small
            red dot (in addition to the curve).
        dpi: Output image resolution in dots per inch.

    """
    fig, ax = plt.subplots(figsize=(10, 10))
    ax.set_aspect("equal")
    cmap_rooms = plt.get_cmap("Pastel1")

    cls._draw_room_floors(ax, rooms, cmap_rooms)
    cls._draw_occupied_area(ax, occ_area)
    cls._draw_footprint_outlines(ax, footprints)
    cls._draw_footprint_labels(ax, footprints)
    cls._draw_room_labels(ax, rooms)
    if trajectory is not None and len(trajectory) > 1:
        cls._draw_trajectory(
            ax,
            np.asarray(trajectory),
            arrow_stride,
            current_index,
            point_markers,
        )
    cls._configure_axes(ax, rooms, occ_area)

    ax.set_title("")
    ax.set_xlabel("")
    ax.set_ylabel("")
    ax.set_xticks([])
    ax.set_yticks([])
    for spine in ax.spines.values():
        spine.set_visible(False)
    fig.subplots_adjust(left=0, right=1, top=1, bottom=0)
    fig.patch.set_alpha(0)
    ax.patch.set_alpha(0)
    plt.savefig(
        save_path,
        dpi=dpi,
        bbox_inches="tight",
        pad_inches=0,
        transparent=True,
    )
    plt.close(fig)

embodied_gen.skills.spatial-computing.core.trajectory

DEFAULT_CLEARANCE module-attribute

DEFAULT_CLEARANCE = 0.4

Minimum distance (m) required between the trajectory and any obstacle.

DEFAULT_ENDPOINT_CLEARANCE module-attribute

DEFAULT_ENDPOINT_CLEARANCE = 1.5

Minimum distance (m) the start/end keep from walls and objects.

DEFAULT_NUM_WAYPOINTS module-attribute

DEFAULT_NUM_WAYPOINTS = 8

Number of roam waypoints sampled across the navigable space.

DEFAULT_OBSTACLE_IGNORE module-attribute

DEFAULT_OBSTACLE_IGNORE = ('door',)

Footprint name keywords excluded from obstacles (treated as open).

DEFAULT_POINT_SPACING module-attribute

DEFAULT_POINT_SPACING = 0.1

Output spacing (m) between consecutive trajectory points.

DEFAULT_RESOLUTION module-attribute

DEFAULT_RESOLUTION = 0.05

Occupancy-grid cell size in meters.

DEFAULT_TURN_RADIUS module-attribute

DEFAULT_TURN_RADIUS = 0.5

Target turning-arc radius (m) for rounding corners.

HAIRPIN_REVERSAL_DOT module-attribute

HAIRPIN_REVERSAL_DOT = -0.85

In/out direction dot product below which a turn is treated as a hairpin (near 180 deg) and collapsed; normal roam turns are preserved.

MIN_FILLET_RADIUS module-attribute

MIN_FILLET_RADIUS = 0.15

Smallest fillet radius (m) tried when rounding a corner; below this a corner that still cannot be rounded is left sharp.

TURN_FRAME_STEP_DEG module-attribute

TURN_FRAME_STEP_DEG = 20.0

Heading step (deg) of each inserted in-place rotation frame.

TURN_FRAME_TRIGGER_DEG module-attribute

TURN_FRAME_TRIGGER_DEG = 45.0

Only heading jumps larger than this (a big U-turn at an unavoidable sharp corner) get extra in-place rotation frames; gentler turns flow continuously.

RoamTrajectoryGenerator

RoamTrajectoryGenerator(floor: Geometry, obstacles: Geometry, clearance: float = DEFAULT_CLEARANCE, resolution: float = DEFAULT_RESOLUTION, rooms: dict[str, Geometry] | None = None)

Generate smooth, collision-free roaming trajectories on a floorplan.

The drivable region is the floor area minus furniture footprints. An occupancy grid and its distance transform give per-cell clearance; cells with clearance >= clearance form the navigable space. Spread waypoints are sampled within one connected navigable component, ordered by polar angle into a non-self-intersecting loop, and linked with a clearance-aware A* planner. Each segment is simplified inside the navigable region, dead- end hairpins are removed, corners are rounded, and a final clearance pass keeps every sample collision-free; the start and end are pushed into open space.

Doors are treated as open passages by default (excluded from obstacles), matching scenes where doors are removed before rendering.

Initialize the generator and build the navigability grid.

Parameters:

Name Type Description Default
floor Geometry

Drivable floor region (e.g. union of all room floors).

required
obstacles Geometry

Union of obstacle footprints (e.g. furniture).

required
clearance float

Minimum obstacle clearance (m) the path must keep.

DEFAULT_CLEARANCE
resolution float

Occupancy-grid cell size in meters.

DEFAULT_RESOLUTION
rooms dict[str, Geometry] | None

Optional room-name to polygon map, used only to report which rooms the trajectory can reach.

None
Source code in embodied_gen/skills/spatial-computing/core/trajectory.py
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def __init__(
    self,
    floor: Geometry,
    obstacles: Geometry,
    clearance: float = DEFAULT_CLEARANCE,
    resolution: float = DEFAULT_RESOLUTION,
    rooms: dict[str, Geometry] | None = None,
) -> None:
    """Initialize the generator and build the navigability grid.

    Args:
        floor: Drivable floor region (e.g. union of all room floors).
        obstacles: Union of obstacle footprints (e.g. furniture).
        clearance: Minimum obstacle clearance (m) the path must keep.
        resolution: Occupancy-grid cell size in meters.
        rooms: Optional room-name to polygon map, used only to report
            which rooms the trajectory can reach.
    """
    if floor.is_empty:
        raise ValueError("Floor region is empty; cannot plan a path.")

    self.clearance = clearance
    self.resolution = resolution
    self.rooms = rooms or {}

    self.minx, self.miny, maxx, maxy = floor.bounds
    self.nx = max(1, int(math.ceil((maxx - self.minx) / resolution)))
    self.ny = max(1, int(math.ceil((maxy - self.miny) / resolution)))

    xs = self.minx + (np.arange(self.nx) + 0.5) * resolution
    ys = self.miny + (np.arange(self.ny) + 0.5) * resolution
    grid_x, grid_y = np.meshgrid(xs, ys)
    flat_x, flat_y = grid_x.ravel(), grid_y.ravel()

    inside = shapely.contains_xy(floor, flat_x, flat_y)
    if obstacles is not None and not obstacles.is_empty:
        in_obs = shapely.contains_xy(obstacles, flat_x, flat_y)
    else:
        in_obs = np.zeros_like(inside)
    free = (inside & ~in_obs).reshape(self.ny, self.nx)

    self.clearance_map = distance_transform_edt(free) * resolution
    self.nav = self.clearance_map >= clearance
    self.labels, n_comp = label(self.nav, structure=np.ones((3, 3)))
    logger.info(
        "Navigability grid %dx%d, %d component(s), max clearance %.2fm.",
        self.ny,
        self.nx,
        n_comp,
        float(self.clearance_map.max()),
    )
from_collector classmethod
from_collector(collector, clearance: float = DEFAULT_CLEARANCE, resolution: float = DEFAULT_RESOLUTION, obstacle_ignore: tuple[str, ...] = DEFAULT_OBSTACLE_IGNORE, obstacle_clearance: float | None = None) -> 'RoamTrajectoryGenerator'

Build a generator from a UrdfSemanticInfoCollector.

Parameters:

Name Type Description Default
collector

A collector that has already parsed a URDF scene.

required
clearance float

Minimum clearance (m) kept from walls (the floor boundary); keep it small so narrow doorways stay passable.

DEFAULT_CLEARANCE
resolution float

Occupancy-grid cell size in meters.

DEFAULT_RESOLUTION
obstacle_ignore tuple[str, ...]

Footprint name keywords to exclude from obstacles (treated as open passages, e.g. doors).

DEFAULT_OBSTACLE_IGNORE
obstacle_clearance float | None

Minimum clearance (m) kept from furniture and objects. When larger than clearance, footprints are inflated by the difference so the path stays this far from them while still only needing clearance from walls. Defaults to clearance (no extra furniture margin).

None

Returns:

Type Description
'RoamTrajectoryGenerator'

A configured RoamTrajectoryGenerator.

Source code in embodied_gen/skills/spatial-computing/core/trajectory.py
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
@classmethod
def from_collector(
    cls,
    collector,
    clearance: float = DEFAULT_CLEARANCE,
    resolution: float = DEFAULT_RESOLUTION,
    obstacle_ignore: tuple[str, ...] = DEFAULT_OBSTACLE_IGNORE,
    obstacle_clearance: float | None = None,
) -> "RoamTrajectoryGenerator":
    """Build a generator from a ``UrdfSemanticInfoCollector``.

    Args:
        collector: A collector that has already parsed a URDF scene.
        clearance: Minimum clearance (m) kept from walls (the floor
            boundary); keep it small so narrow doorways stay passable.
        resolution: Occupancy-grid cell size in meters.
        obstacle_ignore: Footprint name keywords to exclude from
            obstacles (treated as open passages, e.g. doors).
        obstacle_clearance: Minimum clearance (m) kept from furniture and
            objects. When larger than ``clearance``, footprints are
            inflated by the difference so the path stays this far from
            them while still only needing ``clearance`` from walls.
            Defaults to ``clearance`` (no extra furniture margin).

    Returns:
        A configured ``RoamTrajectoryGenerator``.
    """
    obstacle_polys = [
        poly
        for key, poly in collector.footprints.items()
        if not any(kw in key.lower() for kw in obstacle_ignore)
    ]
    obstacles = (
        unary_union(obstacle_polys) if obstacle_polys else Polygon()
    )
    margin = (obstacle_clearance or clearance) - clearance
    if margin > 0 and not obstacles.is_empty:
        obstacles = obstacles.buffer(margin)
    return cls(
        floor=collector.floor_union,
        obstacles=obstacles,
        clearance=clearance,
        resolution=resolution,
        rooms=collector.rooms,
    )
generate
generate(start_xy: tuple[float, float] | None = None, num_waypoints: int = DEFAULT_NUM_WAYPOINTS, point_spacing: float = DEFAULT_POINT_SPACING, turn_radius: float = DEFAULT_TURN_RADIUS, endpoint_clearance: float = DEFAULT_ENDPOINT_CLEARANCE, seed: int | None = None) -> TrajectoryResult

Generate a smooth roaming trajectory.

Parameters:

Name Type Description Default
start_xy tuple[float, float] | None

Optional world (x, y) start. Defaults to a point in the largest navigable component.

None
num_waypoints int

Number of roam waypoints to visit.

DEFAULT_NUM_WAYPOINTS
point_spacing float

Output sampling spacing in meters.

DEFAULT_POINT_SPACING
turn_radius float

Target turning-arc radius (m); larger gives wider, rounder turns (clamped by segment length and clearance).

DEFAULT_TURN_RADIUS
endpoint_clearance float

Minimum clearance (m) kept at the start and end points so they are not placed close to walls.

DEFAULT_ENDPOINT_CLEARANCE
seed int | None

Optional RNG seed for reproducible roaming.

None

Returns:

Type Description
TrajectoryResult

A TrajectoryResult with (x, y, rot_deg) waypoints.

Raises:

Type Description
ValueError

If no navigable path can be planned.

Source code in embodied_gen/skills/spatial-computing/core/trajectory.py
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
def generate(
    self,
    start_xy: tuple[float, float] | None = None,
    num_waypoints: int = DEFAULT_NUM_WAYPOINTS,
    point_spacing: float = DEFAULT_POINT_SPACING,
    turn_radius: float = DEFAULT_TURN_RADIUS,
    endpoint_clearance: float = DEFAULT_ENDPOINT_CLEARANCE,
    seed: int | None = None,
) -> TrajectoryResult:
    """Generate a smooth roaming trajectory.

    Args:
        start_xy: Optional world (x, y) start. Defaults to a point in the
            largest navigable component.
        num_waypoints: Number of roam waypoints to visit.
        point_spacing: Output sampling spacing in meters.
        turn_radius: Target turning-arc radius (m); larger gives wider,
            rounder turns (clamped by segment length and clearance).
        endpoint_clearance: Minimum clearance (m) kept at the start and
            end points so they are not placed close to walls.
        seed: Optional RNG seed for reproducible roaming.

    Returns:
        A ``TrajectoryResult`` with (x, y, rot_deg) waypoints.

    Raises:
        ValueError: If no navigable path can be planned.
    """
    rng = np.random.default_rng(seed)
    mask, comp_cells = self._select_component(start_xy)

    waypoints = self._sample_waypoints(
        comp_cells, num_waypoints, start_xy, rng
    )
    waypoints = self._order_waypoints(waypoints)

    # Keep the end (and the auto-chosen start) well away from walls.
    waypoints[-1] = self._snap_to_open(
        waypoints[-1], mask, endpoint_clearance
    )
    if start_xy is None:
        waypoints[0] = self._snap_to_open(
            waypoints[0], mask, endpoint_clearance
        )

    # Plan each waypoint-to-waypoint segment and simplify it within the
    # navigable mask. Simplifying per segment (not the merged path) keeps
    # every chord inside walls AND preserves the waypoints (roam shape),
    # while collapsing straight runs so corners can use the full radius.
    vertices: list[tuple[float, float]] = []
    for a, b in zip(waypoints[:-1], waypoints[1:]):
        segment = self._astar(a, b, mask)
        if segment is None:
            logger.warning("No A* path between %s and %s; skipping.", a, b)
            continue
        seg_world = [self._to_world(i, j) for i, j in segment]
        seg_world = self._simplify_in_mask(seg_world, mask)
        if vertices and seg_world and vertices[-1] == seg_world[0]:
            seg_world = seg_world[1:]
        vertices.extend(seg_world)

    if len(vertices) < 2:
        raise ValueError(
            "Failed to plan a roaming path; try more waypoints or a "
            "lower clearance."
        )

    # Collapse dead-end hairpins (single-door room excursions) that the
    # waypoint-angle prune cannot detect.
    vertices = self._remove_reversals(vertices, mask)

    world_path = self._round_corners(vertices, turn_radius)
    # Resample to uniform spacing, then enforce clearance last so the
    # output is collision-free (the final interpolation cannot reintroduce
    # violations). Spacing stays near-uniform; only the few points pushed
    # out near tight corners shift, and gen_trajectory derives timestamps
    # from arc length so constant speed is preserved.
    xy = self._resample(world_path, point_spacing)
    xy = self._enforce_clearance(xy)

    rots = self._compute_headings(xy)
    points = np.column_stack([xy, rots])
    # Insert in-place rotations at unavoidably sharp corners so the
    # heading never snaps between consecutive points.
    points = self._limit_heading_rate(points)

    xy = points[:, :2]
    seg = np.linalg.norm(np.diff(xy, axis=0), axis=1)
    length = float(seg.sum())
    min_clr = min(self._clearance_at(x, y) for x, y in xy)
    reachable = self._reachable_rooms(mask)

    logger.info(
        "Generated trajectory: %d points, %.2fm long, min clearance "
        "%.2fm, reaches %d room(s).",
        len(points),
        length,
        min_clr,
        len(reachable),
    )
    return TrajectoryResult(
        points=points,
        clearance=self.clearance,
        min_clearance=min_clr,
        length=length,
        reachable_rooms=reachable,
    )

TrajectoryResult dataclass

TrajectoryResult(points: ndarray, clearance: float, min_clearance: float, length: float, reachable_rooms: list[str] = list())

Result of a roaming-trajectory generation.

Attributes:

Name Type Description
points ndarray

Array (N, 3) of (x, y, rot_deg). rot_deg is the heading tangent to the curve: 0 deg points +Y (12 o'clock), increasing counter-clockwise in [0, 360).

clearance float

Clearance radius (m) used for planning.

min_clearance float

Minimum obstacle clearance (m) along the trajectory.

length float

Total arc length (m) of the trajectory.

reachable_rooms list[str]

Room keys reachable within the planning component.

heading_to_rot_deg

heading_to_rot_deg(dx: float, dy: float) -> float

Convert a motion direction to the roaming heading convention.

The heading is the forward (tangent) direction of travel, where 0 deg points to +Y (12 o'clock) and the angle increases counter-clockwise.

Parameters:

Name Type Description Default
dx float

X component of the motion direction.

required
dy float

Y component of the motion direction.

required

Returns:

Type Description
float

Heading in degrees within [0, 360).

Source code in embodied_gen/skills/spatial-computing/core/trajectory.py
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def heading_to_rot_deg(dx: float, dy: float) -> float:
    """Convert a motion direction to the roaming heading convention.

    The heading is the forward (tangent) direction of travel, where 0 deg
    points to +Y (12 o'clock) and the angle increases counter-clockwise.

    Args:
        dx: X component of the motion direction.
        dy: Y component of the motion direction.

    Returns:
        Heading in degrees within [0, 360).
    """
    return math.degrees(math.atan2(-dx, dy)) % 360.0

embodied_gen.skills.spatial-computing.api.floorplan_api

FloorplanConfig dataclass

FloorplanConfig(urdf_path: str, output_path: str | None = None, usd_path: str | None = None, asset_path: str | None = None, instance_key: str = 'inserted_object', in_room: str | None = None, on_instance: str | None = None, beside_instance: str | None = None, beside_distance: float = DEFAULT_BESIDE_DISTANCE, place_strategy: Literal['top', 'random'] = 'random', rotation_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY, ignore_items: list[str] = (lambda: list(DEFAULT_IGNORE_ITEMS))(), mesh_sample_num: int = DEFAULT_MESH_SAMPLE_NUM, max_placement_attempts: int = 2000, update_urdf: bool = True, update_usd: bool = True, list_instances: bool = False, delete_instance: str | None = None, delete_in_room: str | None = None, query_instance: str | None = None, output_strategy: Literal['suffix', 'overwrite', 'timestamp'] = 'suffix', batch_insert_config: str | None = None)

Configuration for floorplan operations.

asset_path class-attribute instance-attribute
asset_path: str | None = None

Optional path to the asset mesh file (.obj).

batch_insert_config class-attribute instance-attribute
batch_insert_config: str | None = None

Path to JSON config file for batch insertion (3-4x faster than multiple CLI calls).

JSON format example: [ { "asset_path": "path/to/chair1.obj", "instance_key": "chair_1", "beside_instance": "table_dining_7178300", "in_room": "dining_room_0_floor" }, { "asset_path": "path/to/chair2.obj", "instance_key": "chair_2", "beside_instance": "table_dining_7178300", "in_room": "dining_room_0_floor" } ]

beside_distance class-attribute instance-attribute
beside_distance: float = DEFAULT_BESIDE_DISTANCE

Max distance (meters) from the target instance for beside placement.

beside_instance class-attribute instance-attribute
beside_instance: str | None = None

Optional instance name to place the asset beside (on floor, near the target).

delete_in_room class-attribute instance-attribute
delete_in_room: str | None = None

Optional room constraint for deletion - only delete if instance is in this room.

delete_instance class-attribute instance-attribute
delete_instance: str | None = None

Optional instance name to delete from the scene (supports fuzzy matching with LLM).

ignore_items class-attribute instance-attribute
ignore_items: list[str] = field(default_factory=lambda: list(DEFAULT_IGNORE_ITEMS))

List of item name patterns to ignore during parsing.

in_room class-attribute instance-attribute
in_room: str | None = None

Optional room name to constrain asset placement.

instance_key class-attribute instance-attribute
instance_key: str = 'inserted_object'

Unique key for the added instance.

list_instances class-attribute instance-attribute
list_instances: bool = False

If True, print instance and room names then exit (no placement/visualization).

max_placement_attempts class-attribute instance-attribute
max_placement_attempts: int = 2000

Maximum attempts for asset placement.

mesh_sample_num class-attribute instance-attribute
mesh_sample_num: int = DEFAULT_MESH_SAMPLE_NUM

Number of points to sample from meshes.

on_instance class-attribute instance-attribute
on_instance: str | None = None

Optional instance name to place the asset on top of (exact key from get_instance_names()).

output_path class-attribute instance-attribute
output_path: str | None = None

Path to save the floorplan visualization image.

output_strategy class-attribute instance-attribute
output_strategy: Literal['suffix', 'overwrite', 'timestamp'] = 'suffix'

File naming strategy for output files.

  • "suffix": Add '_updated' suffix (default, non-destructive)
  • "overwrite": Overwrite original files (use with caution)
  • "timestamp": Add timestamp suffix (e.g., '_20260311_171500')
place_strategy class-attribute instance-attribute
place_strategy: Literal['top', 'random'] = 'random'

Placement strategy for the asset.

query_instance class-attribute instance-attribute
query_instance: str | None = None

Optional instance name to query and return its center coordinates (supports fuzzy matching with LLM).

rotation_rpy class-attribute instance-attribute
rotation_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY

Rotation in roll-pitch-yaw (radians).

update_urdf class-attribute instance-attribute
update_urdf: bool = True

Whether to update and save the URDF file.

update_usd class-attribute instance-attribute
update_usd: bool = True

Whether to update and save the USD file.

urdf_path instance-attribute
urdf_path: str

Path to the input URDF scene file.

usd_path class-attribute instance-attribute
usd_path: str | None = None

Optional path to the USD scene file for USD export.

FloorplanManager

FloorplanManager(urdf_path: str, usd_path: str | None = None, mesh_sample_num: int = DEFAULT_MESH_SAMPLE_NUM, ignore_items: list[str] | None = None, output_strategy: Literal['suffix', 'overwrite', 'timestamp'] = 'suffix')

High-level API for floorplan operations.

This class provides simplified methods for: - Loading and analyzing URDF scenes - Visualizing floorplans - Inserting objects into scenes - Updating URDF and USD files

Example

manager = FloorplanManager(urdf_path="scene.urdf", usd_path="scene.usdc") manager.visualize(output_path="floorplan.png") position = manager.insert_object( ... asset_path="chair.obj", ... instance_key="chair_1", ... in_room="kitchen" ... )

URDF/USD are updated automatically after insert

Initialize the floorplan manager.

Parameters:

Name Type Description Default
urdf_path str

Path to the URDF file.

required
usd_path str | None

Optional path to the USD file for scene updates.

None
mesh_sample_num int

Number of points to sample from meshes.

DEFAULT_MESH_SAMPLE_NUM
ignore_items list[str] | None

List of item name patterns to ignore.

None
output_strategy Literal['suffix', 'overwrite', 'timestamp']

File naming strategy for output files.

'suffix'
Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def __init__(
    self,
    urdf_path: str,
    usd_path: str | None = None,
    mesh_sample_num: int = DEFAULT_MESH_SAMPLE_NUM,
    ignore_items: list[str] | None = None,
    output_strategy: Literal[
        "suffix", "overwrite", "timestamp"
    ] = "suffix",
) -> None:
    """Initialize the floorplan manager.

    Args:
        urdf_path: Path to the URDF file.
        usd_path: Optional path to the USD file for scene updates.
        mesh_sample_num: Number of points to sample from meshes.
        ignore_items: List of item name patterns to ignore.
        output_strategy: File naming strategy for output files.

    """
    self.urdf_path = urdf_path
    self.usd_path = usd_path
    self.output_strategy = output_strategy
    self.collector = UrdfSemanticInfoCollector(
        mesh_sample_num=mesh_sample_num,
        ignore_items=ignore_items,
    )
    self.collector.collect(urdf_path)
    self.pending_instance_data: dict | None = None
batch_insert_objects
batch_insert_objects(objects: list[dict], defer_update: bool = False) -> list[list[float] | None]

Batch insert multiple objects into the scene efficiently.

Parameters:

Name Type Description Default
objects list[dict]

List of object configs, each containing: asset_path: Path to the asset mesh file (.obj). instance_key: Unique key for the new instance. in_room: Optional room name to constrain placement. on_instance: Optional instance name to place on top of. beside_instance: Optional instance name to place beside. beside_distance: Max distance from target (default: 0.5m). rotation_rpy: Initial rotation (default: (0, 0, 0)). place_strategy: Either "top" or "random" (default: "random").

required
defer_update bool

If True, don't update URDF/USD after each insertion. Useful when inserting many objects at once.

False

Returns:

Type Description
list[list[float] | None]

List of centers [x, y, z] for each inserted object,

list[list[float] | None]

or None if failed.

Example

objects = [ ... {"asset_path": "chair1.obj", ... "instance_key": "chair_1", ... "beside_instance": "table"}, ... ] centers = manager.batch_insert_objects(objects)

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
def batch_insert_objects(
    self,
    objects: list[dict],
    defer_update: bool = False,
) -> list[list[float] | None]:
    """Batch insert multiple objects into the scene efficiently.

    Args:
        objects: List of object configs, each containing:
            asset_path: Path to the asset mesh file (.obj).
            instance_key: Unique key for the new instance.
            in_room: Optional room name to constrain placement.
            on_instance: Optional instance name to place on top of.
            beside_instance: Optional instance name to place beside.
            beside_distance: Max distance from target (default: 0.5m).
            rotation_rpy: Initial rotation (default: (0, 0, 0)).
            place_strategy: Either "top" or "random" (default: "random").
        defer_update: If True, don't update URDF/USD after each
            insertion. Useful when inserting many objects at once.

    Returns:
        List of centers [x, y, z] for each inserted object,
        or None if failed.

    Example:
        >>> objects = [
        ...     {"asset_path": "chair1.obj",
        ...      "instance_key": "chair_1",
        ...      "beside_instance": "table"},
        ... ]
        >>> centers = manager.batch_insert_objects(objects)

    """
    centers = []
    usd_source = self.usd_path

    for i, obj_config in enumerate(objects, 1):
        logger.info(
            f"[{i}/{len(objects)}] Inserting '{obj_config.get('instance_key', 'unknown')}'..."
        )

        center = self.collector.add_instance(
            asset_path=obj_config["asset_path"],
            instance_key=obj_config["instance_key"],
            in_room=obj_config.get("in_room"),
            on_instance=obj_config.get("on_instance"),
            beside_instance=obj_config.get("beside_instance"),
            beside_distance=obj_config.get(
                "beside_distance", DEFAULT_BESIDE_DISTANCE
            ),
            rotation_rpy=obj_config.get(
                "rotation_rpy", DEFAULT_ROTATION_RPY
            ),
            n_max_attempt=obj_config.get("n_max_attempt", 2000),
            place_strategy=obj_config.get("place_strategy", "random"),
        )

        if center is not None:
            # Store instance data for later update
            collision_path = obj_config["asset_path"].replace(
                ".obj", "_collision.obj"
            )
            if not os.path.exists(collision_path):
                collision_path = None

            # Update URDF incrementally
            if self.urdf_path and not defer_update:
                urdf_out = self._get_output_path(self.urdf_path)
                self.collector.update_urdf_info(
                    output_path=urdf_out,
                    instance_key=obj_config["instance_key"],
                    visual_mesh_path=obj_config["asset_path"],
                    collision_mesh_path=collision_path,
                    trans_xyz=tuple(center),
                    rot_rpy=obj_config.get(
                        "rotation_rpy", DEFAULT_ROTATION_RPY
                    ),
                    joint_type="fixed",
                )

            # Update USD incrementally
            if self.usd_path and not defer_update:
                usd_out = self._get_output_path(self.usd_path)
                self.collector.update_usd_info(
                    usd_path=usd_source,
                    output_path=usd_out,
                    instance_key=obj_config["instance_key"],
                    visual_mesh_path=obj_config["asset_path"],
                    trans_xyz=center,
                    rot_rpy=obj_config.get(
                        "rotation_rpy", DEFAULT_ROTATION_RPY
                    ),
                )
                usd_source = usd_out

            logger.info(f"✅ Placed at {center}")
        else:
            logger.warning(f"❌ Failed to place object")

        centers.append(center)

    return centers
delete_object
delete_object(instance_key: str, in_room: str | None = None, urdf_output_path: str | None = None, usd_output_path: str | None = None) -> bool

Delete an object from the scene.

Parameters:

Name Type Description Default
instance_key str

Exact instance name to delete.

required
in_room str | None

Optional room constraint - only delete if instance is in this room.

None
urdf_output_path str | None

Optional custom path for URDF output.

None
usd_output_path str | None

Optional custom path for USD output.

None

Returns:

Type Description
bool

True if deletion succeeded, False otherwise.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
def delete_object(
    self,
    instance_key: str,
    in_room: str | None = None,
    urdf_output_path: str | None = None,
    usd_output_path: str | None = None,
) -> bool:
    """Delete an object from the scene.

    Args:
        instance_key: Exact instance name to delete.
        in_room: Optional room constraint - only delete if instance is in this room.
        urdf_output_path: Optional custom path for URDF output.
        usd_output_path: Optional custom path for USD output.

    Returns:
        True if deletion succeeded, False otherwise.

    """
    success = self.collector.remove_instance(
        instance_key=instance_key,
        in_room=in_room,
    )

    if success:
        # Update URDF
        if self.urdf_path:
            urdf_out = self._get_output_path(
                self.urdf_path, urdf_output_path
            )
            self.collector.save_urdf(urdf_out)

        # Update USD
        if self.usd_path:
            usd_out = self._get_output_path(self.usd_path, usd_output_path)
            self.collector.remove_usd_instance(
                usd_path=self.usd_path,
                output_path=usd_out,
                instance_key=instance_key,
            )

    return success
get_floor_union
get_floor_union() -> Geometry

Get the union of all floor areas.

Returns:

Type Description
Geometry

Shapely geometry representing floor areas.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
791
792
793
794
795
796
797
798
def get_floor_union(self) -> Geometry:
    """Get the union of all floor areas.

    Returns:
        Shapely geometry representing floor areas.

    """
    return self.collector.floor_union
get_instance_names
get_instance_names() -> list[str]

Get list of instance names in the scene.

Returns:

Type Description
list[str]

List of instance names.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
500
501
502
503
504
505
506
507
508
509
510
511
def get_instance_names(self) -> list[str]:
    """Get list of instance names in the scene.

    Returns:
        List of instance names.

    """
    return [
        k
        for k in self.collector.instances.keys()
        if k != "walls" and "floor" not in k.lower()
    ]
get_instance_names_in_room
get_instance_names_in_room(in_room: str) -> list[str]

Get instance names that are spatially inside a given room.

Buffers the room polygon slightly to handle mesh-sampling precision.

Parameters:

Name Type Description Default
in_room str

Exact room key (must exist in get_room_names()).

required

Returns:

Type Description
list[str]

List of instance names within the room.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
def get_instance_names_in_room(self, in_room: str) -> list[str]:
    """Get instance names that are spatially inside a given room.

    Buffers the room polygon slightly to handle mesh-sampling precision.

    Args:
        in_room: Exact room key (must exist in get_room_names()).

    Returns:
        List of instance names within the room.

    """
    room_poly = self.collector.rooms.get(in_room)
    if room_poly is None:
        return self.get_instance_names()
    room_buffered = room_poly.buffer(0.1)
    all_names = self.get_instance_names()
    return [
        k
        for k in all_names
        if room_buffered.contains(
            self.collector.instances[k].representative_point()
        )
    ]
get_occupied_area
get_occupied_area() -> Geometry

Get the union of all occupied areas.

Returns:

Type Description
Geometry

Shapely geometry representing occupied areas.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
782
783
784
785
786
787
788
789
def get_occupied_area(self) -> Geometry:
    """Get the union of all occupied areas.

    Returns:
        Shapely geometry representing occupied areas.

    """
    return self.collector.occ_area
get_room_names
get_room_names() -> list[str]

Get list of room names in the scene.

Returns:

Type Description
list[str]

List of room names.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
513
514
515
516
517
518
519
520
def get_room_names(self) -> list[str]:
    """Get list of room names in the scene.

    Returns:
        List of room names.

    """
    return list(self.collector.rooms.keys())
insert_object
insert_object(asset_path: str, instance_key: str, in_room: str | None = None, on_instance: str | None = None, beside_instance: str | None = None, beside_distance: float = DEFAULT_BESIDE_DISTANCE, rotation_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY, n_max_attempt: int = 2000, place_strategy: Literal['top', 'random'] = 'random') -> list[float] | None

Insert an object into the scene with automatic placement.

Parameters:

Name Type Description Default
asset_path str

Path to the asset mesh file (.obj).

required
instance_key str

Unique key for the new instance.

required
in_room str | None

Optional room name to constrain placement.

None
on_instance str | None

Optional instance name to place on top of.

None
beside_instance str | None

Optional instance name to place beside (on floor).

None
beside_distance float

Max distance from target for beside placement.

DEFAULT_BESIDE_DISTANCE
rotation_rpy tuple[float, float, float]

Initial rotation in roll-pitch-yaw.

DEFAULT_ROTATION_RPY
n_max_attempt int

Maximum placement attempts.

2000
place_strategy Literal['top', 'random']

Either "top" or "random".

'random'

Returns:

Type Description
list[float] | None

List [x, y, z] of the placed instance center, or None if failed.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def insert_object(
    self,
    asset_path: str,
    instance_key: str,
    in_room: str | None = None,
    on_instance: str | None = None,
    beside_instance: str | None = None,
    beside_distance: float = DEFAULT_BESIDE_DISTANCE,
    rotation_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY,
    n_max_attempt: int = 2000,
    place_strategy: Literal["top", "random"] = "random",
) -> list[float] | None:
    """Insert an object into the scene with automatic placement.

    Args:
        asset_path: Path to the asset mesh file (.obj).
        instance_key: Unique key for the new instance.
        in_room: Optional room name to constrain placement.
        on_instance: Optional instance name to place on top of.
        beside_instance: Optional instance name to place beside (on floor).
        beside_distance: Max distance from target for beside placement.
        rotation_rpy: Initial rotation in roll-pitch-yaw.
        n_max_attempt: Maximum placement attempts.
        place_strategy: Either "top" or "random".

    Returns:
        List [x, y, z] of the placed instance center, or None if failed.

    """
    center = self.collector.add_instance(
        asset_path=asset_path,
        instance_key=instance_key,
        in_room=in_room,
        on_instance=on_instance,
        beside_instance=beside_instance,
        beside_distance=beside_distance,
        rotation_rpy=rotation_rpy,
        n_max_attempt=n_max_attempt,
        place_strategy=place_strategy,
    )

    if center is not None:
        self.pending_instance_data = {
            "asset_path": asset_path,
            "instance_key": instance_key,
            "center": center,
            "rotation_rpy": rotation_rpy,
        }
        self.update_scene()

    return center
query_instance_center
query_instance_center(instance_key: str) -> list[float] | None

Query the center coordinates of an instance.

Parameters:

Name Type Description Default
instance_key str

Exact instance name to query.

required

Returns:

Type Description
list[float] | None

List [x, y, z] of the instance center, or None if not found.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
711
712
713
714
715
716
717
718
719
720
721
722
723
724
def query_instance_center(
    self,
    instance_key: str,
) -> list[float] | None:
    """Query the center coordinates of an instance.

    Args:
        instance_key: Exact instance name to query.

    Returns:
        List [x, y, z] of the instance center, or None if not found.

    """
    return self.collector.get_instance_center(instance_key)
resolve_and_query_instance
resolve_and_query_instance(query_instance: str, gpt_client: object | None = None) -> tuple[str | None, list[float] | None]

Resolve instance name and return its center coordinates.

Combines fuzzy matching with coordinate query. If query_instance is already in get_instance_names(), return its center. Otherwise if gpt_client is provided, use LLM to resolve user description (e.g. "床", "沙发") to one exact instance key.

Parameters:

Name Type Description Default
query_instance str

Exact instance key or semantic description.

required
gpt_client object | None

Optional GPT client for semantic resolve.

None

Returns:

Type Description
tuple[str | None, list[float] | None]

Tuple of (resolved_instance_name, center_coordinates), or (None, None) if not found.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
def resolve_and_query_instance(
    self,
    query_instance: str,
    gpt_client: object | None = None,
) -> tuple[str | None, list[float] | None]:
    r"""Resolve instance name and return its center coordinates.

    Combines fuzzy matching with coordinate query.
    If query_instance is already in get_instance_names(), return its center.
    Otherwise if gpt_client is provided, use LLM to resolve user description
    (e.g. "床", "沙发") to one exact instance key.

    Args:
        query_instance: Exact instance key or semantic description.
        gpt_client: Optional GPT client for semantic resolve.

    Returns:
        Tuple of (resolved_instance_name, center_coordinates), or (None, None) if not found.

    """
    names = self.get_instance_names()

    if query_instance in names:
        center = self.query_instance_center(query_instance)
        return query_instance, center

    # Substring matching as fallback
    query_lower = query_instance.lower()
    matches = [n for n in names if query_lower in n.lower()]
    if len(matches) == 1:
        logger.info(
            "Substring match: '%s' -> '%s'", query_instance, matches[0]
        )
        center = self.query_instance_center(matches[0])
        return matches[0], center
    elif len(matches) > 1:
        logger.warning(
            "Multiple substring matches for '%s': %s. Using '%s'. "
            "Use exact name or LLM for better matching.",
            query_instance,
            matches,
            matches[0],
        )
        center = self.query_instance_center(matches[0])
        return matches[0], center

    if gpt_client is not None:
        resolved = resolve_instance_with_llm(
            gpt_client, names, query_instance  # type: ignore[arg-type]
        )
        if resolved:
            center = self.query_instance_center(resolved)
            return resolved, center

    return None, None
resolve_beside_instance
resolve_beside_instance(beside_instance: str, gpt_client: object | None = None, in_room: str | None = None) -> str | None

Resolve beside_instance to an exact key (for beside placement).

If beside_instance is already in get_instance_names(), return it. Otherwise if gpt_client is provided, use LLM to resolve user description (e.g. "桌子", "沙发") to one exact instance key.

When in_room is given, only instances spatially inside that room are considered as candidates.

Parameters:

Name Type Description Default
beside_instance str

Exact instance key or semantic description.

required
gpt_client object | None

Optional GPT client for semantic resolve.

None
in_room str | None

Optional resolved room key to restrict candidate scope.

None

Returns:

Type Description
str | None

Exact instance key, or None if not found / LLM returned NONE.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
def resolve_beside_instance(
    self,
    beside_instance: str,
    gpt_client: object | None = None,
    in_room: str | None = None,
) -> str | None:
    r"""Resolve beside_instance to an exact key (for beside placement).

    If beside_instance is already in get_instance_names(), return it.
    Otherwise if gpt_client is provided, use LLM to resolve user description
    (e.g. "桌子", "沙发") to one exact instance key.

    When `in_room` is given, only instances spatially inside that room are
    considered as candidates.

    Args:
        beside_instance: Exact instance key or semantic description.
        gpt_client: Optional GPT client for semantic resolve.
        in_room: Optional resolved room key to restrict candidate scope.

    Returns:
        Exact instance key, or None if not found / LLM returned NONE.
    """
    if in_room is not None:
        names = self.get_instance_names_in_room(in_room)
    else:
        names = self.get_instance_names()
    if beside_instance in names:
        return beside_instance

    # Substring matching as fallback
    query_lower = beside_instance.lower()
    matches = [n for n in names if query_lower in n.lower()]
    if len(matches) == 1:
        logger.info(
            "Substring match: '%s' -> '%s'", beside_instance, matches[0]
        )
        return matches[0]
    elif len(matches) > 1:
        logger.warning(
            "Multiple substring matches for '%s': %s. Using '%s'. "
            "Use exact name or LLM for better matching.",
            beside_instance,
            matches,
            matches[0],
        )
        return matches[0]

    if gpt_client is not None:
        return resolve_instance_with_llm(
            gpt_client, names, beside_instance  # type: ignore[arg-type]
        )
    return None
resolve_delete_instance
resolve_delete_instance(delete_instance: str, gpt_client: object | None = None, in_room: str | None = None) -> str | None

Resolve delete_instance to an exact key (for deletion).

Similar to resolve_beside_instance but specifically for deletion. If delete_instance is already in get_instance_names(), return it. Otherwise if gpt_client is provided, use LLM to resolve user description (e.g. "桌子", "沙发") to one exact instance key.

When in_room is given, only instances spatially inside that room are considered as candidates.

Parameters:

Name Type Description Default
delete_instance str

Exact instance key or semantic description.

required
gpt_client object | None

Optional GPT client for semantic resolve.

None
in_room str | None

Optional resolved room key to restrict candidate scope.

None

Returns:

Type Description
str | None

Exact instance key, or None if not found / LLM returned NONE.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
def resolve_delete_instance(
    self,
    delete_instance: str,
    gpt_client: object | None = None,
    in_room: str | None = None,
) -> str | None:
    r"""Resolve delete_instance to an exact key (for deletion).

    Similar to resolve_beside_instance but specifically for deletion.
    If delete_instance is already in get_instance_names(), return it.
    Otherwise if gpt_client is provided, use LLM to resolve user description
    (e.g. "桌子", "沙发") to one exact instance key.

    When `in_room` is given, only instances spatially inside that room are
    considered as candidates.

    Args:
        delete_instance: Exact instance key or semantic description.
        gpt_client: Optional GPT client for semantic resolve.
        in_room: Optional resolved room key to restrict candidate scope.

    Returns:
        Exact instance key, or None if not found / LLM returned NONE.
    """
    if in_room is not None:
        names = self.get_instance_names_in_room(in_room)
    else:
        names = self.get_instance_names()

    if delete_instance in names:
        return delete_instance

    # Substring matching as fallback
    query_lower = delete_instance.lower()
    matches = [n for n in names if query_lower in n.lower()]
    if len(matches) == 1:
        logger.info(
            "Substring match: '%s' -> '%s'", delete_instance, matches[0]
        )
        return matches[0]
    elif len(matches) > 1:
        logger.warning(
            "Multiple substring matches for '%s': %s. Using '%s'. "
            "Use exact name or LLM for better matching.",
            delete_instance,
            matches,
            matches[0],
        )
        return matches[0]

    if gpt_client is not None:
        return resolve_instance_with_llm(
            gpt_client, names, delete_instance  # type: ignore[arg-type]
        )
    return None
resolve_in_room
resolve_in_room(in_room: str, gpt_client: object | None = None) -> str | None

Resolve in_room to an exact room name (for placement).

If in_room is already in get_room_names(), return it. Otherwise if gpt_client is provided, use LLM to resolve user description (e.g. \"kitchen\", \"the place for cooking\") to one exact room name.

Parameters:

Name Type Description Default
in_room str

Exact room name or semantic description.

required
gpt_client object | None

Optional GPT client for semantic resolve (e.g. GPT_CLIENT).

None

Returns:

Type Description
str | None

Exact room name, or None if not found / LLM returned NONE.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
def resolve_in_room(
    self,
    in_room: str,
    gpt_client: object | None = None,
) -> str | None:
    r"""Resolve in_room to an exact room name (for placement).

    If in_room is already in get_room_names(), return it.
    Otherwise if gpt_client is provided, use LLM to resolve user description
    (e.g. \"kitchen\", \"the place for cooking\") to one exact room name.

    Args:
        in_room: Exact room name or semantic description.
        gpt_client: Optional GPT client for semantic resolve (e.g. GPT_CLIENT).

    Returns:
        Exact room name, or None if not found / LLM returned NONE.
    """
    names = self.get_room_names()
    if in_room in names:
        return in_room
    if gpt_client is not None:
        return resolve_instance_with_llm(
            gpt_client, names, in_room  # type: ignore[arg-type]
        )
    return None
resolve_on_instance
resolve_on_instance(on_instance: str, gpt_client: object | None = None) -> str | None

Resolve on_instance to an exact key (for placement).

If on_instance is already in get_instance_names(), return it. Otherwise if gpt_client is provided, use LLM to resolve user description (e.g. \"柜子\", \"书柜\") to one exact instance key.

Parameters:

Name Type Description Default
on_instance str

Exact instance key or semantic description.

required
gpt_client object | None

Optional GPT client for semantic resolve (e.g. GPT_CLIENT).

None

Returns:

Type Description
str | None

Exact instance key, or None if not found / LLM returned NONE.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
def resolve_on_instance(
    self,
    on_instance: str,
    gpt_client: object | None = None,
) -> str | None:
    r"""Resolve on_instance to an exact key (for placement).

    If on_instance is already in get_instance_names(), return it.
    Otherwise if gpt_client is provided, use LLM to resolve user description
    (e.g. \"柜子\", \"书柜\") to one exact instance key.

    Args:
        on_instance: Exact instance key or semantic description.
        gpt_client: Optional GPT client for semantic resolve (e.g. GPT_CLIENT).

    Returns:
        Exact instance key, or None if not found / LLM returned NONE.
    """
    names = self.get_instance_names()
    if on_instance in names:
        return on_instance
    if gpt_client is not None:
        return resolve_instance_with_llm(
            gpt_client, names, on_instance  # type: ignore[arg-type]
        )
    return None
update_scene
update_scene(urdf_output_path: str | None = None, usd_output_path: str | None = None) -> None

Update URDF and/or USD with inserted instances.

Updates URDF if self.urdf_path is set, USD if self.usd_path is set. Both are updated when both paths are set. No-op when no instance was inserted.

Note: USD updates require Blender (bpy) to convert .obj to .usdc format.

Parameters:

Name Type Description Default
urdf_output_path str | None

Optional custom path for URDF output.

None
usd_output_path str | None

Optional custom path for USD output.

None

Raises:

Type Description
ValueError

If no instance has been inserted.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
def update_scene(
    self,
    urdf_output_path: str | None = None,
    usd_output_path: str | None = None,
) -> None:
    """Update URDF and/or USD with inserted instances.

    Updates URDF if self.urdf_path is set, USD if self.usd_path is set.
    Both are updated when both paths are set. No-op when no instance was inserted.

    Note: USD updates require Blender (bpy) to convert .obj to .usdc format.

    Args:
        urdf_output_path: Optional custom path for URDF output.
        usd_output_path: Optional custom path for USD output.

    Raises:
        ValueError: If no instance has been inserted.

    """
    if self.pending_instance_data is None:
        raise ValueError(
            "No instance to update. Call insert_object() first."
        )

    data = self.pending_instance_data
    collision_path = data["asset_path"].replace(".obj", "_collision.obj")
    if not os.path.exists(collision_path):
        collision_path = None

    if self.urdf_path:
        urdf_out = self._get_output_path(self.urdf_path, urdf_output_path)
        self.collector.update_urdf_info(
            output_path=urdf_out,
            instance_key=data["instance_key"],
            visual_mesh_path=data["asset_path"],
            collision_mesh_path=collision_path,
            trans_xyz=tuple(data["center"]),
            rot_rpy=data["rotation_rpy"],
            joint_type="fixed",
        )

    if self.usd_path:
        usd_out = self._get_output_path(self.usd_path, usd_output_path)
        self.collector.update_usd_info(
            usd_path=self.usd_path,
            output_path=usd_out,
            instance_key=data["instance_key"],
            visual_mesh_path=data["asset_path"],
            trans_xyz=data["center"],
            rot_rpy=data["rotation_rpy"],
        )
visualize
visualize(output_path: str) -> None

Generate and save a floorplan visualization.

Parameters:

Name Type Description Default
output_path str

Path to save the output image.

required
Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def visualize(
    self,
    output_path: str,
) -> None:
    """Generate and save a floorplan visualization.

    Args:
        output_path: Path to save the output image.

    """
    FloorplanVisualizer.plot(
        self.collector.rooms,
        self.collector.footprints,
        self.collector.occ_area,
        output_path,
    )
    logger.info(f"✅ Floorplan visualization saved to {output_path}")

delete_object_from_scene

delete_object_from_scene(urdf_path: str, instance_key: str, in_room: str | None = None, output_path: str | None = None) -> bool

Quick function to delete an object from scene.

Parameters:

Name Type Description Default
urdf_path str

Path to the URDF file.

required
instance_key str

Exact instance name to delete.

required
in_room str | None

Optional room constraint - only delete if instance is in this room.

None
output_path str | None

Optional path to save the floorplan image after deletion.

None

Returns:

Type Description
bool

True if deletion succeeded, False otherwise.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
def delete_object_from_scene(
    urdf_path: str,
    instance_key: str,
    in_room: str | None = None,
    output_path: str | None = None,
) -> bool:
    """Quick function to delete an object from scene.

    Args:
        urdf_path: Path to the URDF file.
        instance_key: Exact instance name to delete.
        in_room: Optional room constraint - only delete if instance is in this room.
        output_path: Optional path to save the floorplan image after deletion.

    Returns:
        True if deletion succeeded, False otherwise.

    """
    manager = FloorplanManager(urdf_path=urdf_path)
    success = manager.delete_object(
        instance_key=instance_key,
        in_room=in_room,
    )
    if success and output_path is not None:
        manager.visualize(output_path=output_path)
    return success

insert_object_to_scene

insert_object_to_scene(urdf_path: str, asset_path: str, instance_key: str, output_path: str, usd_path: str | None = None, in_room: str | None = None, on_instance: str | None = None, beside_instance: str | None = None, beside_distance: float = DEFAULT_BESIDE_DISTANCE, place_strategy: Literal['top', 'random'] = 'random', rotation_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY) -> list[float] | None

Quick function to insert an object and generate floorplan.

Note: USD updates require Blender (bpy) to convert .obj to .usdc format.

Parameters:

Name Type Description Default
urdf_path str

Path to the URDF file.

required
asset_path str

Path to the asset mesh file (.obj).

required
instance_key str

Unique key for the new instance.

required
output_path str

Path to save the floorplan image.

required
usd_path str | None

Optional path to the USD file (requires Blender).

None
in_room str | None

Optional room name to constrain placement.

None
on_instance str | None

Optional instance name to place on top of.

None
beside_instance str | None

Optional instance name to place beside (on floor).

None
beside_distance float

Max distance for beside placement (meters).

DEFAULT_BESIDE_DISTANCE
place_strategy Literal['top', 'random']

Either "top" or "random".

'random'
rotation_rpy tuple[float, float, float]

Initial rotation in roll-pitch-yaw.

DEFAULT_ROTATION_RPY

Returns:

Type Description
list[float] | None

List [x, y, z] of the placed instance center, or None if failed.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
def insert_object_to_scene(
    urdf_path: str,
    asset_path: str,
    instance_key: str,
    output_path: str,
    usd_path: str | None = None,
    in_room: str | None = None,
    on_instance: str | None = None,
    beside_instance: str | None = None,
    beside_distance: float = DEFAULT_BESIDE_DISTANCE,
    place_strategy: Literal["top", "random"] = "random",
    rotation_rpy: tuple[float, float, float] = DEFAULT_ROTATION_RPY,
) -> list[float] | None:
    """Quick function to insert an object and generate floorplan.

    Note: USD updates require Blender (bpy) to convert .obj to .usdc format.

    Args:
        urdf_path: Path to the URDF file.
        asset_path: Path to the asset mesh file (.obj).
        instance_key: Unique key for the new instance.
        output_path: Path to save the floorplan image.
        usd_path: Optional path to the USD file (requires Blender).
        in_room: Optional room name to constrain placement.
        on_instance: Optional instance name to place on top of.
        beside_instance: Optional instance name to place beside (on floor).
        beside_distance: Max distance for beside placement (meters).
        place_strategy: Either "top" or "random".
        rotation_rpy: Initial rotation in roll-pitch-yaw.

    Returns:
        List [x, y, z] of the placed instance center, or None if failed.

    """
    manager = FloorplanManager(urdf_path=urdf_path, usd_path=usd_path)
    center = manager.insert_object(
        asset_path=asset_path,
        instance_key=instance_key,
        in_room=in_room,
        on_instance=on_instance,
        beside_instance=beside_instance,
        beside_distance=beside_distance,
        rotation_rpy=rotation_rpy,
        place_strategy=place_strategy,
    )
    if center is not None:
        manager.visualize(output_path=output_path)
    return center

query_instance_position

query_instance_position(urdf_path: str, instance_key: str) -> list[float] | None

Quick function to query instance center coordinates.

Parameters:

Name Type Description Default
urdf_path str

Path to the URDF file.

required
instance_key str

Exact instance name to query.

required

Returns:

Type Description
list[float] | None

List [x, y, z] of the instance center, or None if not found.

Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
def query_instance_position(
    urdf_path: str,
    instance_key: str,
) -> list[float] | None:
    """Quick function to query instance center coordinates.

    Args:
        urdf_path: Path to the URDF file.
        instance_key: Exact instance name to query.

    Returns:
        List [x, y, z] of the instance center, or None if not found.

    """
    manager = FloorplanManager(urdf_path=urdf_path)
    return manager.query_instance_center(instance_key)

visualize_floorplan

visualize_floorplan(urdf_path: str, output_path: str, mesh_sample_num: int = DEFAULT_MESH_SAMPLE_NUM, ignore_items: list[str] | None = None) -> None

Quick function to visualize a floorplan.

Parameters:

Name Type Description Default
urdf_path str

Path to the URDF file.

required
output_path str

Path to save the output image.

required
mesh_sample_num int

Number of points to sample from meshes.

DEFAULT_MESH_SAMPLE_NUM
ignore_items list[str] | None

List of item name patterns to ignore.

None
Source code in embodied_gen/skills/spatial-computing/api/floorplan_api.py
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
def visualize_floorplan(
    urdf_path: str,
    output_path: str,
    mesh_sample_num: int = DEFAULT_MESH_SAMPLE_NUM,
    ignore_items: list[str] | None = None,
) -> None:
    """Quick function to visualize a floorplan.

    Args:
        urdf_path: Path to the URDF file.
        output_path: Path to save the output image.
        mesh_sample_num: Number of points to sample from meshes.
        ignore_items: List of item name patterns to ignore.

    """
    manager = FloorplanManager(
        urdf_path=urdf_path,
        mesh_sample_num=mesh_sample_num,
        ignore_items=ignore_items,
    )
    manager.visualize(output_path=output_path)