robolib.serialization

entites.py Helpers to make various entities, like cubes, robots, meshes Import level: 3

 1"""
 2entites.py Helpers to make various entities, like cubes, robots, meshes
 3Import level: 3
 4"""
 5from pathlib import Path
 6from typing import Any
 7import numpy as np
 8from microecs import World, EntityId, ComponentType, Entity
 9from robolib.utils import RlModelWithTexture, get_project_root, make_radius
10from robolib.components import HasModel, HasPose, HasFPV, ROBOLIB_COMPONENTS
11
12def world_to_dict(world: World) -> dict[str, Any]:
13    """Serialize the world. Goes through all the entities and their components and converts the serializables to dict"""
14    res = {"entities": [], "components": world.component_names, "extra_metadata": world.extra_metadata}
15    for eid in world.live_entities.keys():
16        res["entities"].append({"entity_id": eid, **world.get_entity(eid).to_dict(serialization_field="serializable")})
17    return res
18
19def world_from_dict(data: dict[str, Any]) -> World:
20    """Creates a world from a serialized representation e.g. from world_to_dict()"""
21    components = [{c.__name__: c for c in ROBOLIB_COMPONENTS}[c] for c in data["components"]]
22    world = World(components=components, extra_metadata=data["extra_metadata"])
23    for entity in data["entities"]:
24        components = [world.component_name_to_type[c] for c in entity["components"]]
25        add_entity(world, components, **entity["data"])
26    world.update()
27    return world
28
29def make_nonserializable_data(components: list[ComponentType], data: dict[str, Any]) -> dict[str, np.ndarray]:
30    """
31    Given the serializable data (e.g. from entity_to_dict()), create the live non-serializable (e.g. gpu model).
32    This is the missing link that allows us to spawn new entities (via world.add_entity).
33    """
34    res = {}
35    for component in components:
36        if component == HasModel:
37            model = RlModelWithTexture.from_path(data["model_path"], data["texture_path"])
38            top_left, bottom_right = model.bbox
39            res["model"] = np.array([model], "object")
40            res["model_bbox"] = np.float32([top_left, bottom_right])
41            res["model_radius"] = np.float32([make_radius(size=bottom_right - top_left)])
42        elif component == HasPose:
43            res["candidate_pose"] = data["pose"].copy()
44        elif component == HasFPV:
45            res["fpv_camera"] = np.array([None], "object")
46            res["fpv_data"] = np.array([None], "object")
47            res["fpv_texture"] = np.array([None], "object")
48    return res
49
50def add_entity(world: World, components: list[ComponentType | str], **kwargs) -> EntityId:
51    """Calls world.make_entity() but only after calling make_nonserializable_data. Bridges robosim with microecs."""
52    components: list[ComponentType] = [world.component_name_to_type[c] if isinstance(c, str) else c for c in components]
53
54    for key in ("model_path", "texture_path"): # store portable, root-relative strings
55        if kwargs.get(key) is not None:
56            p = Path(kwargs[key])
57            kwargs[key] = str(p.relative_to(get_project_root()) if p.is_absolute() else p)
58    return world.add_entity(components, **kwargs, **make_nonserializable_data(components, kwargs))
59
60def add_component(entity: Entity, component: ComponentType, **kwargs):
61    """Small wrapper on top of entity.add_component handling the non serializable data"""
62    entity.add_component(component, **kwargs, **make_nonserializable_data([component], kwargs))
def world_to_dict(world: microecs.world.World) -> dict[str, typing.Any]:
13def world_to_dict(world: World) -> dict[str, Any]:
14    """Serialize the world. Goes through all the entities and their components and converts the serializables to dict"""
15    res = {"entities": [], "components": world.component_names, "extra_metadata": world.extra_metadata}
16    for eid in world.live_entities.keys():
17        res["entities"].append({"entity_id": eid, **world.get_entity(eid).to_dict(serialization_field="serializable")})
18    return res

Serialize the world. Goes through all the entities and their components and converts the serializables to dict

def world_from_dict(data: dict[str, typing.Any]) -> microecs.world.World:
20def world_from_dict(data: dict[str, Any]) -> World:
21    """Creates a world from a serialized representation e.g. from world_to_dict()"""
22    components = [{c.__name__: c for c in ROBOLIB_COMPONENTS}[c] for c in data["components"]]
23    world = World(components=components, extra_metadata=data["extra_metadata"])
24    for entity in data["entities"]:
25        components = [world.component_name_to_type[c] for c in entity["components"]]
26        add_entity(world, components, **entity["data"])
27    world.update()
28    return world

Creates a world from a serialized representation e.g. from world_to_dict()

def make_nonserializable_data( components: list[type[microecs.component.Component]], data: dict[str, typing.Any]) -> dict[str, numpy.ndarray]:
30def make_nonserializable_data(components: list[ComponentType], data: dict[str, Any]) -> dict[str, np.ndarray]:
31    """
32    Given the serializable data (e.g. from entity_to_dict()), create the live non-serializable (e.g. gpu model).
33    This is the missing link that allows us to spawn new entities (via world.add_entity).
34    """
35    res = {}
36    for component in components:
37        if component == HasModel:
38            model = RlModelWithTexture.from_path(data["model_path"], data["texture_path"])
39            top_left, bottom_right = model.bbox
40            res["model"] = np.array([model], "object")
41            res["model_bbox"] = np.float32([top_left, bottom_right])
42            res["model_radius"] = np.float32([make_radius(size=bottom_right - top_left)])
43        elif component == HasPose:
44            res["candidate_pose"] = data["pose"].copy()
45        elif component == HasFPV:
46            res["fpv_camera"] = np.array([None], "object")
47            res["fpv_data"] = np.array([None], "object")
48            res["fpv_texture"] = np.array([None], "object")
49    return res

Given the serializable data (e.g. from entity_to_dict()), create the live non-serializable (e.g. gpu model). This is the missing link that allows us to spawn new entities (via world.add_entity).

def add_entity( world: microecs.world.World, components: list[type[microecs.component.Component] | str], **kwargs) -> int:
51def add_entity(world: World, components: list[ComponentType | str], **kwargs) -> EntityId:
52    """Calls world.make_entity() but only after calling make_nonserializable_data. Bridges robosim with microecs."""
53    components: list[ComponentType] = [world.component_name_to_type[c] if isinstance(c, str) else c for c in components]
54
55    for key in ("model_path", "texture_path"): # store portable, root-relative strings
56        if kwargs.get(key) is not None:
57            p = Path(kwargs[key])
58            kwargs[key] = str(p.relative_to(get_project_root()) if p.is_absolute() else p)
59    return world.add_entity(components, **kwargs, **make_nonserializable_data(components, kwargs))

Calls world.make_entity() but only after calling make_nonserializable_data. Bridges robosim with microecs.

def add_component( entity: microecs.entity.Entity, component: type[microecs.component.Component], **kwargs):
61def add_component(entity: Entity, component: ComponentType, **kwargs):
62    """Small wrapper on top of entity.add_component handling the non serializable data"""
63    entity.add_component(component, **kwargs, **make_nonserializable_data([component], kwargs))

Small wrapper on top of entity.add_component handling the non serializable data