OmniSimulator API Reference

This section provides comprehensive API documentation for the OmniSimulator text-based simulation engine.

Core Engine

SimulationEngine

class OmniSimulator.core.engine.SimulationEngine(config: Dict[str, Any] | None = None, scene_abilities: List[str] = None)[source]

Bases: object

模拟引擎类 - 整个模拟器的核心控制器

__init__(config=None)[source]

Initialize the simulation engine with optional configuration.

Parameters:: config (dict or None) – Configuration dictionary or None for defaults

initialize_with_task(task_file)[source]

Initialize simulation with a specific task file.

Parameters:: task_file (str) – Path to task JSON file
Returns:: True if initialization successful
Return type:: bool

process_command(command)[source]

Process a text command and return the result.

Parameters:: command (str) – Natural language command
Returns:: Command execution result
Return type:: dict

__init__(config: Dict[str, Any] | None = None, scene_abilities: List[str] = None)[source]

初始化模拟引擎

Parameters:

config – 全局配置字典，可选
scene_abilities – 场景特定的能力列表，用于动态注册需要工具的动作

initialize(scene_file: str, agent_file: str | None = None) → bool[source]

初始化模拟器 :param scene_file: 场景文件路径 :param agent_file: 智能体文件路径（已废弃，不再使用）

Returns:: 是否成功初始化
Return type:: bool

initialize_with_task(task_file: str) → bool[source]

使用任务文件初始化模拟器（包括场景和智能体）

Parameters:: task_file – 任务文件路径
Returns:: 是否成功初始化
Return type:: bool

initialize_with_data(data: Dict[str, Any]) → bool[source]

使用数据字典初始化模拟器

Parameters:

data –

包含场景、任务和动作配置的数据字典格式: {

’scene’: scene_data, # 场景数据 ‘task’: task_data, # 任务数据 (可选) ‘actions’: action_config_data # 动作配置数据 (可选)

}

Returns:

是否成功初始化

Return type:

bool

validate_environment() → Tuple[bool, List[str]][source]

验证当前环境是否合法

Returns:: (是否合法, 错误信息列表)
Return type:: Tuple[bool, List[str]]

load_agents(agents_config: List[Dict[str, Any]]) → bool[source]

加载智能体配置

Parameters:: agents_config – 智能体配置列表
Returns:: 是否成功加载
Return type:: bool

process_command(agent_id: str, command: str)[source]

处理智能体的命令

Parameters:

agent_id – 智能体ID
command – 命令字符串

Returns:

(执行状态, 反馈消息, 结果数据)

Return type:

Tuple

get_agent_info(agent_id: str) → Dict[str, Any] | None[source]

获取智能体信息

Parameters:: agent_id – 智能体ID
Returns:: 智能体信息字典
Return type:: Dict

get_visualization_status() → Dict[str, Any][source]

获取可视化系统状态

Returns:: 可视化系统状态信息
Return type:: Dict

get_visualization_url() → str | None[source]

获取可视化Web界面URL

Returns:: 可视化URL，如果未启用则返回None
Return type:: str

stop_visualization()[source]: 停止可视化系统

restart_visualization() → bool[source]

重启可视化系统

Returns:: 是否成功重启
Return type:: bool

get_object_info(object_id: str) → Dict[str, Any] | None[source]

获取物体信息

Parameters:: object_id – 物体ID
Returns:: 物体信息字典
Return type:: Dict

get_room_info(room_id: str) → Dict[str, Any] | None[source]

获取房间信息

Parameters:: room_id – 房间ID
Returns:: 房间信息字典
Return type:: Dict

get_task_info() → Dict[str, Any] | None[source]

获取当前任务信息

Returns:: 任务信息字典
Return type:: Dict

update_scene_abilities(new_abilities: List[str])[source]

更新场景能力并重新注册相关动作

Parameters:: new_abilities – 新的能力列表

get_task_verification_status() → Dict[str, Any] | None[source]

获取任务验证状态

Returns:: 验证状态，如果未启用验证则返回None
Return type:: Optional[Dict[str, Any]]

set_task_data(task_data: Dict[str, Any])[source]

设置任务数据并创建验证器

Parameters:: task_data – 任务数据，来自task.json文件

get_agent_supported_actions_description(agent_ids: List[str]) → str[source]

获取智能体支持的所有动作的字符串描述

Parameters:: agent_ids – 智能体ID列表，支持单个或多个智能体
Returns:: 包含所有支持动作的描述字符串（英文）
Return type:: str

Environment Management

EnvironmentManager

class OmniSimulator.environment.environment_manager.EnvironmentManager(world_state: WorldState, sim_config: Dict[str, Any] | None = None)[source]

Bases: object

环境管理器 - 负责管理模拟环境中的所有实体（房间、物体、家具）

get_room(room_id)

Retrieve room information by ID.

Parameters:: room_id (str) – Unique room identifier
Returns:: Room object or None if not found
Return type:: Room or None

__init__(world_state: WorldState, sim_config: Dict[str, Any] | None = None)[source]

初始化环境管理器

Parameters:

world_state – 世界状态对象
sim_config – 全局配置字典，可选

load_scene(scene_data: Dict[str, Any]) → bool[source]

从场景数据加载环境

Parameters:: scene_data – 场景数据字典
Returns:: 加载是否成功
Return type:: bool

load_scene_old(scene_data: Dict[str, Any]) → bool[source]

从场景数据加载环境

Parameters:: scene_data – 场景数据字典
Returns:: 加载是否成功
Return type:: bool

add_room(room: Room) → bool[source]

添加房间到环境

Parameters:: room – 房间对象
Returns:: 添加是否成功
Return type:: bool

parse_location_id(location_id: str)[source]

add_object(obj_data: Dict[str, Any], location_id: str) → str | None[source]

添加物体到环境

Parameters:

obj_data – 物体数据字典
location_id – 物体位置ID (房间ID或容器物体ID)

Returns:

添加成功则返回物体ID，否则返回None

Return type:

str

connect_rooms(room_id1: str, room_id2: str) → bool[source]

连接两个房间

Parameters:

room_id1 – 第一个房间ID
room_id2 – 第二个房间ID

Returns:

连接是否成功

Return type:

bool

get_room_by_id(room_id: str) → Dict[str, Any] | None[source]

获取房间数据

Parameters:: room_id – 房间ID
Returns:: 房间数据字典，如果不存在则返回None
Return type:: Dict

get_object_by_id(object_id: str) → Dict[str, Any] | None[source]

获取物体数据

Parameters:: object_id – 物体ID
Returns:: 物体数据字典，如果不存在则返回None
Return type:: Dict

get_objects_in_room(room_id: str, recursive: bool = True) → List[Dict[str, Any]][source]

获取房间中的所有物体

Parameters:

room_id – 房间ID
recursive – 是否递归获取容器中的物体

Returns:

物体数据字典列表

Return type:

List[Dict]

get_discovered_objects_in_room(room_id: str, recursive: bool = True) → List[Dict[str, Any]][source]

获取房间中的所有已发现物体

Parameters:

room_id – 房间ID
recursive – 是否递归获取容器中的物体

Returns:

已发现物体数据字典列表

Return type:

List[Dict]

update_object_state(object_id: str, state_updates: Dict[str, Any]) → bool[source]

更新物体状态

Parameters:

object_id – 物体ID
state_updates – 状态更新字典

Returns:

更新是否成功

Return type:

bool

discover_object(object_id: str) → bool[source]

将物体标记为已发现

Parameters:: object_id – 物体ID
Returns:: 更新是否成功
Return type:: bool

discover_objects_in_room(room_id: str, percentage: float = 1.0) → List[str][source]

随机发现房间中的某个百分比的物体

Parameters:

room_id – 房间ID
percentage – 要发现的物体百分比 (0.0-1.0)

Returns:

新发现的物体ID列表

Return type:

List[str]

move_object(object_id: str, new_location_id: str) → bool[source]: 移动物体到新位置（location_id始终为房间ID，on/in关系通过图结构维护）

get_object_location(object_id: str) → Dict[str, Any] | None[source]

获取物体位置

Parameters:: object_id – 物体ID
Returns:: 位置数据字典，如果物体不存在则返回None
Return type:: Dict

find_path(start_room_id: str, end_room_id: str) → List[str] | None[source]

查找从一个房间到另一个房间的路径

Parameters:

start_room_id – 起始房间ID
end_room_id – 目标房间ID

Returns:

路径房间ID列表，如果不存在路径则返回None

Return type:

List[str]

is_object_accessible(object_id: str, agent_id: str) → Tuple[bool, str][source]

检查物体是否对智能体可访问

Parameters:

object_id – 物体ID
agent_id – 智能体ID

Returns:

(是否可访问, 原因消息)

Return type:

Tuple[bool, str]

get_object_room(object_id: str) → str | None[source]

获取物体所在的房间ID

Parameters:: object_id – 物体ID
Returns:: 房间ID，如果物体不存在或不在任何房间则返回None
Return type:: str

is_agent_near_object(agent_id: str, object_id: str) → Tuple[bool, str][source]

检查智能体是否与物体在同一房间

Parameters:

agent_id – 智能体ID
object_id – 物体ID

Returns:

(是否在同一房间, 消息)

Return type:

Tuple[bool, str]

update_object_attributes(object_id: str, updates: Dict[str, Any]) → bool[source]: 通用物体属性更新方法，可更新任意顶层字段（如 holders, location_id 等）

Room

class OmniSimulator.environment.room.Room(room_id: str, name: str, properties: Dict[str, Any] | None = None)[source]

Bases: object

房间类 - 表示模拟环境中的房间

room_id: str: Unique identifier for the room.

name: str: Human-readable room name.

get_objects_in_room(room_id)

Get all objects currently in the specified room.

Parameters:: room_id (str) – Room identifier
Returns:: List of objects in the room
Return type:: list

move_object(object_id, target_location)

Move an object to a new location.

Parameters:

object_id (str) – Object to move
target_location (str) – Destination location

Returns:

True if move successful

Return type:

bool

__init__(room_id: str, name: str, properties: Dict[str, Any] | None = None)[source]

初始化房间

Parameters:

room_id – 房间唯一ID
name – 房间名称
properties – 房间属性字典，如 {‘size’: ‘large’, ‘type’: ‘kitchen’}

connect_to(room_id: str) → None[source]

连接到另一个房间

Parameters:: room_id – 要连接的房间ID

disconnect_from(room_id: str) → None[source]

断开与另一个房间的连接

Parameters:: room_id – 要断开连接的房间ID

is_connected_to(room_id: str) → bool[source]

检查是否连接到特定房间

Parameters:: room_id – 要检查的房间ID
Returns:: 是否连接
Return type:: bool

to_dict() → Dict[str, Any][source]: 将房间对象转换为字典表示

classmethod from_dict(data: Dict[str, Any]) → Room[source]: 从字典创建房间对象

Action System

ActionManager

class OmniSimulator.action.action_manager.ActionManager(world_state, env_manager, agent_manager, scene_abilities=None)[source]

Bases: object

动作管理器 - 负责动作的创建、验证和执行

这个类整合了原有的ActionParser, ActionValidator和ActionExecutor的功能，采用了面向对象的设计模式，使代码更清晰、易于维护和扩展。

主要职责： 1. 解析命令字符串，创建动作对象 2. 验证动作是否可以执行 3. 执行动作并返回结果 4. 维护协作动作的状态

register_action(action_class)

Register a new action class with the system.

Parameters:: action_class (class) – Action class to register

execute_action(agent, action_type, **kwargs)[source]

Execute an action for the specified agent.

Parameters:

agent (Agent) – Agent performing the action
action_type (str) – Type of action to perform

Returns:

Action execution result

Return type:

dict

validate_action(agent, action_type, **kwargs)[source]

Validate if an action can be performed.

Parameters:

agent (Agent) – Agent attempting the action
action_type (str) – Type of action to validate

Returns:

Validation result

Return type:

dict

agent_action_classes = {}

__init__(world_state, env_manager, agent_manager, scene_abilities=None)[source]

初始化动作管理器

Parameters:

world_state – 世界状态对象
env_manager – 环境管理器
agent_manager – 智能体管理器
scene_abilities – 场景支持的能力列表，用于注册不需要工具的属性动作

classmethod register_ability_action(action_name: str, agent_id: str)[source]

为特定智能体注册能力相关动作（仅限需要工具的动作）

Parameters:

action_name – 动作名称
agent_id – 智能体ID

classmethod unregister_ability_action(action_name: str, agent_id: str)[source]

为特定智能体解绑能力相关动作（仅限需要工具的动作）

Parameters:

action_name – 动作名称
agent_id – 智能体ID

register_action_class(action_name: str, action_class: Type[BaseAction])[source]

注册单个动作类

Parameters:

action_name – 动作名称，将自动转为大写
action_class – 动作类，必须是BaseAction的子类

Returns:

支持链式调用

Return type:

self

register_action_classes(action_classes_dict: Dict[str, Type[BaseAction]])[source]

批量注册多个动作类

Parameters:: action_classes_dict – 动作名称到动作类的映射字典
Returns:: 支持链式调用
Return type:: self

is_action_registered(action_name: str) → bool[source]

检查动作是否已注册

Parameters:: action_name – 动作名称
Returns:: 是否已注册
Return type:: bool

parse_command(command_str: str, agent_id: str) → Tuple[BaseAction | None, str | None][source]

解析命令字符串并创建相应的动作对象

Parameters:

command_str – 命令字符串，如”GRAB cup_1”
agent_id – 执行动作的智能体ID

Returns:

(动作对象, 错误消息): 如果解析成功，错误消息为None；如果解析失败，动作对象为None

Return type:

Tuple[Optional[BaseAction], Optional[str]]

validate_action(action: BaseAction) → Tuple[bool, str][source]

验证动作是否有效

Parameters:: action – 动作对象
Returns:: (是否有效, 原因消息)
Return type:: Tuple[bool, str]

execute_action(action: BaseAction) → Tuple[ActionStatus, str, Dict[str, Any] | None][source]

执行动作

Parameters:: action – 动作对象
Returns:: (执行状态, 反馈消息, 额外结果数据)
Return type:: Tuple[ActionStatus, str, Optional[Dict]]

process_command(command_str: str, agent_id: str) → Tuple[ActionStatus, str, Dict[str, Any] | None][source]

处理命令字符串，包括解析、验证和执行

Parameters:

command_str – 命令字符串
agent_id – 执行命令的智能体ID

Returns:

(执行状态, 反馈消息, 额外结果数据)

Return type:

Tuple[ActionStatus, str, Optional[Dict]]

get_agent_supported_actions_description(agent_ids: List[str]) → str[source]

获取智能体支持的所有动作的字符串描述

Parameters:: agent_ids – 智能体ID列表，支持单个或多个智能体
Returns:: 包含所有支持动作的描述字符串（英文）
Return type:: str

BaseAction

class OmniSimulator.action.actions.base_action.BaseAction(agent_id: str, action_type: ActionType | None = None, target_id: str | None = None, params: Dict[str, Any] | None = None)[source]

Bases: object

动作基类 - 所有具体动作类的父类

实现动作的核心功能： 1. 从命令字符串解析生成对应动作 2. 验证动作是否可执行 3. 执行动作并返回结果

abstractmethod validate()[source]

Validate if the action can be performed.

Returns:: True if action is valid
Return type:: bool

abstractmethod execute()[source]

Execute the action.

Returns:: Action execution result
Return type:: dict

command_pattern: ClassVar[str] = None

__init__(agent_id: str, action_type: ActionType | None = None, target_id: str | None = None, params: Dict[str, Any] | None = None)[source]

初始化动作

Parameters:

agent_id – 执行动作的智能体ID
action_type – 动作类型，可选（如果不提供则使用类级别的action_type）
target_id – 动作的目标对象ID，可选
params – 动作的附加参数，可选

action_type: ClassVar[ActionType] = None

classmethod from_command(command_str: str, agent_id: str) → BaseAction | None[source]

从命令字符串创建动作实例

Parameters:

command_str – 命令字符串，如”GRAB cup_1”
agent_id – 执行动作的智能体ID

Returns:

成功则返回动作实例，失败则返回None

Return type:

BaseAction

validate(world_state, env_manager, agent_manager) → Tuple[bool, str][source]

验证动作是否可执行

Parameters:

world_state – 世界状态
env_manager – 环境管理器
agent_manager – 智能体管理器

Returns:

(是否有效, 原因消息)

Return type:

Tuple[bool, str]

execute(world_state, env_manager, agent_manager) → Tuple[ActionStatus, str, Dict[str, Any] | None][source]

执行动作

Parameters:

world_state – 世界状态
env_manager – 环境管理器
agent_manager – 智能体管理器

Returns:

(执行状态, 反馈消息, 额外结果数据)

Return type:

Tuple[ActionStatus, str, Optional[Dict]]

to_dict() → Dict[str, Any][source]: 将动作转换为字典表示

Action Types

Basic Actions

class OmniSimulator.action.actions.basic_actions.GotoAction(agent_id: str, action_type: ActionType | None = None, target_id: str | None = None, params: Dict[str, Any] | None = None)[source]

Bases: BaseAction

GOTO动作 - 移动到指定位置

action_type: ClassVar[ActionType] = 1

command_pattern: ClassVar[str] = '^GOTO\\s+(\\w+)$'

class OmniSimulator.action.actions.basic_actions.GrabAction(agent_id: str, action_type: ActionType | None = None, target_id: str | None = None, params: Dict[str, Any] | None = None)[source]

Bases: BaseAction

GRAB动作 - 抓取物体

action_type: ClassVar[ActionType] = 2

command_pattern: ClassVar[str] = '^GRAB\\s+(\\w+)$'

class OmniSimulator.action.actions.basic_actions.PlaceAction(agent_id: str, action_type: ActionType | None = None, target_id: str | None = None, params: Dict[str, Any] | None = None)[source]

Bases: BaseAction

PLACE动作 - 放置物体

action_type: ClassVar[ActionType] = 3

command_pattern: ClassVar[str] = '^PLACE\\s+(\\w+)(?:\\s+(on|in)\\s+(\\w+))?$'

class OmniSimulator.action.actions.attribute_actions.AttributeAction(agent_id: str, action_type: ActionType | None = None, target_id: str | None = None, params: Dict[str, Any] | None = None)[source]

Bases: BaseAction

基于属性的动作 - 从CSV文件自动导入的动作定义

CSV格式: action_name,attribute,value,requires_tool

当物体的属性名=属性值时，可以执行Action操作，且会把属性值取反

两种类型的动作： 1. requires_tool=true：需要智能体拥有相应能力，动态注册 2. requires_tool=false：不需要工具，在初始化时一次性注册

action_type: ClassVar[ActionType] = 9

command_pattern: ClassVar[str] = '^(\\w+)\\s+(\\w+)$'

action_configs = {}

no_tool_actions = {}

classmethod load_from_csv(csv_path=None)[source]: 从CSV文件加载动作配置

classmethod register_scene_no_tool_actions_to_manager(action_manager, scene_abilities)[source]

根据场景abilities注册不需要工具的动作到动作管理器

Parameters:

action_manager – ActionManager实例
scene_abilities – 场景支持的能力列表

Returns:

支持链式调用

Return type:

action_manager

classmethod register_task_specific_actions(action_manager, task_abilities: List[str])[source]

根据任务的abilities动态注册需要工具的动作

Parameters:

action_manager – ActionManager实例
task_abilities – 任务中指定的能力列表

classmethod get_available_actions_for_abilities(abilities: List[str]) → List[str][source]

获取指定能力列表对应的可用动作

Parameters:: abilities – 能力列表
Returns:: 可用的动作名称列表
Return type:: List[str]

classmethod from_command(command_str: str, agent_id: str) → BaseAction | None[source]: 从命令字符串创建动作实例

class OmniSimulator.action.actions.basic_actions.LookAction(agent_id: str, action_type: ActionType | None = None, target_id: str | None = None, params: Dict[str, Any] | None = None)[source]

Bases: BaseAction

LOOK动作 - 观察环境或物体

action_type: ClassVar[ActionType] = 4

command_pattern: ClassVar[str] = '^LOOK(?:\\s+(\\w+))?$'

class OmniSimulator.action.actions.basic_actions.ExploreAction(agent_id: str, action_type: ActionType | None = None, target_id: str | None = None, params: Dict[str, Any] | None = None)[source]

Bases: BaseAction

EXPLORE动作 - 探索当前房间

action_type: ClassVar[ActionType] = 5

command_pattern: ClassVar[str] = '^EXPLORE(?:\\s+(\\w+))?$'

Agent System

Agent

class OmniSimulator.agent.agent.Agent(agent_id: str, name: str, location_id: str, max_grasp_limit: int = 2, properties: Dict[str, Any] | None = None)[source]

Bases: object

智能体类 - 表示模拟环境中的智能体

__init__(agent_id, name, location_id)[source]

Initialize a new agent.

Parameters:

agent_id (str) – Unique agent identifier
name (str) – Human-readable agent name
location_id (str) – Initial location

location_id: str: Current location of the agent.

inventory: list: List of objects currently carried by the agent.

can_grab(object_id)[source]

Check if the agent can grab a specific object.

Parameters:: object_id (str) – Object to check
Returns:: True if object can be grabbed
Return type:: bool

can_carry(object_id)[source]

Check if the agent can carry a specific object.

Parameters:: object_id (str) – Object to check
Returns:: True if object can be carried
Return type:: bool

__init__(agent_id: str, name: str, location_id: str, max_grasp_limit: int = 2, properties: Dict[str, Any] | None = None)[source]

初始化智能体

Parameters:

agent_id – 智能体唯一ID
name – 智能体名称
location_id – 智能体初始位置ID (房间ID)
max_grasp_limit – 最大可抓取物体数量
properties – 其他属性字典，包含max_length/width/height/weight等

can_grab() → bool[source]

检查智能体是否还能抓取更多物体（数量限制）

Returns:: 是否可以抓取
Return type:: bool

can_carry(object_properties: Dict[str, Any]) → Tuple[bool, str][source]

Check if agent can carry the object (weight limit only) Cooperative mode has no weight limit

Parameters:: object_properties – 物体属性字典，包含weight等
Returns:: (是否可以承载, 原因)
Return type:: Tuple[bool, str]

grab_object(object_id: str, object_properties: Dict[str, Any]) → Tuple[bool, str][source]

抓取物体

Parameters:

object_id – 物体ID
object_properties – 物体属性

Returns:

(是否成功抓取, 原因)

Return type:

Tuple[bool, str]

drop_object(object_id: str, object_properties: Dict[str, Any]) → Tuple[bool, str][source]

放下物体

Parameters:

object_id – 物体ID
object_properties – 物体属性

Returns:

(是否成功放下, 原因)

Return type:

Tuple[bool, str]

move_to(new_location_id: str) → None[source]

移动到新位置

Parameters:: new_location_id – 新位置ID

update_near_objects(near_id: str | None = None, env_manager=None)[source]

更新智能体可交互的物体集合

Parameters:

near_id – 靠近的物体ID，如果为None则只保留库存和位置
env_manager – 环境管理器实例，用于查询物体关系

add_ability_from_object(ability: str, object_id: str) → None[source]

从物体获取能力

Parameters:

ability – 能力名称
object_id – 提供能力的物体ID

remove_ability_from_object(ability: str, object_id: str) → None[source]

移除来自特定物体的能力

Parameters:

ability – 能力名称
object_id – 提供能力的物体ID

has_ability(ability: str) → bool[source]

检查是否拥有特定能力

Parameters:: ability – 能力名称
Returns:: 是否拥有该能力
Return type:: bool

to_dict() → Dict[str, Any][source]

将智能体转换为字典表示

Returns:: 智能体数据字典
Return type:: Dict

classmethod from_dict(data: Dict[str, Any]) → Agent[source]

从字典创建智能体对象

Parameters:: data – 智能体数据字典
Returns:: 智能体对象
Return type:: Agent

AgentManager

class OmniSimulator.agent.agent_manager.AgentManager(world_state: WorldState, config: Dict[str, Any] | None = None)[source]

Bases: object

智能体管理器 - 负责管理所有智能体

get_agent(agent_id)[source]

Retrieve an agent by ID.

Parameters:: agent_id (str) – Agent identifier
Returns:: Agent object or None
Return type:: Agent or None

get_all_agents()[source]

Get all active agents.

Returns:: List of all agents
Return type:: list

__init__(world_state: WorldState, config: Dict[str, Any] | None = None)[source]

初始化智能体管理器

Parameters:

world_state – 世界状态对象
config – 全局配置字典，可选

add_agent(agent_data: Dict[str, Any]) → str | None[source]

添加智能体

Parameters:: agent_data – 智能体数据字典
Returns:: 添加成功则返回智能体ID，否则返回None
Return type:: str

get_agent(agent_id: str) → Agent | None[source]

获取智能体实例

Parameters:: agent_id – 智能体ID
Returns:: 智能体实例，如果不存在则返回None
Return type:: Agent

update_agent(agent_id: str, update_data: Dict[str, Any]) → bool[source]

更新智能体数据

Parameters:

agent_id – 智能体ID
update_data – 更新数据字典

Returns:

更新是否成功

Return type:

bool

move_agent(agent_id: str, new_location_id: str) → bool[source]

移动智能体到新位置

Parameters:

agent_id – 智能体ID
new_location_id – 新位置ID

Returns:

移动是否成功

Return type:

bool

add_to_inventory(agent_id: str, object_id: str) → bool[source]

将物体添加到智能体库存

Parameters:

agent_id – 智能体ID
object_id – 物体ID

Returns:

添加是否成功

Return type:

bool

remove_from_inventory(agent_id: str, object_id: str) → bool[source]

从智能体库存移除物体

Parameters:

agent_id – 智能体ID
object_id – 物体ID

Returns:

移除是否成功

Return type:

bool

get_agent_inventory(agent_id: str) → List[str][source]

获取智能体库存

Parameters:: agent_id – 智能体ID
Returns:: 物体ID列表
Return type:: List[str]

get_all_agents() → Dict[str, Agent][source]

获取所有智能体

Returns:: 智能体ID到智能体实例的映射
Return type:: Dict[str, Agent]

Utilities

DataLoader

class OmniSimulator.utils.data_loader.DataLoader(data_dir: str = None)[source]

Bases: object

数据加载器 - 负责加载场景和任务数据

load_task(task_file)[source]

Load task data from file.

Parameters:: task_file (str) – Path to task file
Returns:: Task data dictionary
Return type:: dict

__init__(data_dir: str = None)[source]

初始化数据加载器

Parameters:: data_dir – 数据目录路径，默认为项目根目录下的data文件夹

load_scene(scene_id: str) → Dict[str, Any] | None[source]

加载场景数据

Parameters:: scene_id – 场景ID
Returns:: 场景数据，如果加载失败返回None
Return type:: Dict[str, Any]

load_task(task_id: str) → Dict[str, Any] | None[source]

加载任务数据

Parameters:: task_id – 任务ID
Returns:: 任务数据，如果加载失败返回None
Return type:: Dict[str, Any]

load_complete_scenario(scenario_id: str) → Tuple[Dict, Dict] | None[source]

加载完整的场景数据（场景+任务）

Parameters:: scenario_id – 场景ID
Returns:: (场景数据, 任务数据)，如果任何一个加载失败返回None
Return type:: Tuple[Dict, Dict]

get_available_scenarios() → List[str][source]

获取所有可用的场景ID

Returns:: 可用的场景ID列表
Return type:: List[str]

get_scene_abilities(scene_id: str) → List[str][source]

获取场景中定义的abilities

Parameters:: scene_id – 场景ID
Returns:: 能力列表
Return type:: List[str]

get_agent_configs(task_id: str) → List[Dict[str, Any]][source]

获取任务中定义的智能体配置

Parameters:: task_id – 任务ID
Returns:: 智能体配置列表
Return type:: List[Dict[str, Any]]

get_task_list(task_id: str, task_category: str = None) → List[Dict[str, Any]][source]

获取任务列表

Parameters:

task_id – 任务ID
task_category – 任务类别（如’direct_command’, ‘tool_use’等），如果为None则返回所有任务

Returns:

任务列表

Return type:

List[Dict[str, Any]]

validate_scenario_integrity(scenario_id: str) → bool[source]

验证场景数据的完整性

Parameters:: scenario_id – 场景ID
Returns:: 是否完整
Return type:: bool

Task Verifier

任务验证器 - 负责验证子任务的完成情况

该模块提供了TaskVerifier类，用于根据task.json文件中的validation_checks字段检查子任务是否已完成，并返回详细的验证结果。

class OmniSimulator.utils.task_verifier.TaskVerificationResult(task_id: str, task_description: str)[source]

任务验证结果类

__init__(task_id: str, task_description: str)[source]

mark_completed(details: Dict[str, Any] = None)[source]: 标记任务为已完成

mark_failed(error_message: str)[source]: 标记任务验证失败

to_dict() → Dict[str, Any][source]: 转换为字典格式

class OmniSimulator.utils.task_verifier.TaskVerifier(task_data: Dict[str, Any], env_manager, config: Dict[str, Any] = None)[source]

任务验证器 - 验证子任务完成情况

__init__(task_data: Dict[str, Any], env_manager, config: Dict[str, Any] = None)[source]

初始化任务验证器

Parameters:

task_data – 任务数据，来自task.json文件
env_manager – 环境管理器，用于获取物体状态
config – 验证配置

verify_all_tasks() → List[TaskVerificationResult][source]

验证所有任务

Returns:: 验证结果列表
Return type:: List[TaskVerificationResult]

verify_task_category(category: str) → List[TaskVerificationResult][source]

验证特定类别的任务

Parameters:: category – 任务类别，如 “direct_command”, “tool_use”等
Returns:: 验证结果列表
Return type:: List[TaskVerificationResult]

get_completion_summary() → Dict[str, Any][source]

获取任务完成情况摘要

Returns:: 完成情况摘要
Return type:: Dict[str, Any]

get_subtask_completion_list() → List[bool][source]

获取所有子任务的完成状态列表

Returns:: 按顺序返回每个子任务的完成状态 [True, False, True, …]
Return type:: List[bool]

verify_single_subtask(subtask: Dict[str, Any]) → TaskVerificationResult[source]

验证单个子任务

Parameters:: subtask – 子任务定义，包含验证条件
Returns:: 验证结果
Return type:: TaskVerificationResult

get_current_completion_status() → Dict[str, Any][source]

获取当前所有任务的完成状态

Returns:: 包含完成状态的详细信息
Return type:: Dict[str, Any]

class OmniSimulator.utils.task_verifier.TaskVerifier(task_data: Dict[str, Any], env_manager, config: Dict[str, Any] = None)[source]

Bases: object

任务验证器 - 验证子任务完成情况

__init__(task_data: Dict[str, Any], env_manager, config: Dict[str, Any] = None)[source]

初始化任务验证器

Parameters:

task_data – 任务数据，来自task.json文件
env_manager – 环境管理器，用于获取物体状态
config – 验证配置

verify_all_tasks() → List[TaskVerificationResult][source]

验证所有任务

Returns:: 验证结果列表
Return type:: List[TaskVerificationResult]

verify_task_category(category: str) → List[TaskVerificationResult][source]

验证特定类别的任务

Parameters:: category – 任务类别，如 “direct_command”, “tool_use”等
Returns:: 验证结果列表
Return type:: List[TaskVerificationResult]

get_completion_summary() → Dict[str, Any][source]

获取任务完成情况摘要

Returns:: 完成情况摘要
Return type:: Dict[str, Any]

get_subtask_completion_list() → List[bool][source]

获取所有子任务的完成状态列表

Returns:: 按顺序返回每个子任务的完成状态 [True, False, True, …]
Return type:: List[bool]

verify_single_subtask(subtask: Dict[str, Any]) → TaskVerificationResult[source]

验证单个子任务

Parameters:: subtask – 子任务定义，包含验证条件
Returns:: 验证结果
Return type:: TaskVerificationResult

get_current_completion_status() → Dict[str, Any][source]

获取当前所有任务的完成状态

Returns:: 包含完成状态的详细信息
Return type:: Dict[str, Any]

Core Types and Enums

ObjectType

class OmniSimulator.core.enums.ObjectType(value)[source]

物体类型枚举

STATIC = 1

INTERACTABLE = 2

GRABBABLE = 3

FURNITURE = 4

ITEM = 5

ActionType

class OmniSimulator.core.enums.ActionType(value)[source]

动作类型枚举

GOTO = 1

GRAB = 2

PLACE = 3

LOOK = 4

EXPLORE = 5

CORP_GRAB = 6

CORP_GOTO = 7

CORP_PLACE = 8

ATTRIBUTE = 9

ActionStatus

class OmniSimulator.core.enums.ActionStatus(value)[source]

动作执行状态枚举

SUCCESS = 1

FAILURE = 2

INVALID = 3

PARTIAL = 4

WAITING = 5

Usage Examples

Basic Simulation Setup

from OmniSimulator.core.engine import SimulationEngine

# Initialize simulation engine
engine = SimulationEngine()

# Load a task
success = engine.initialize_with_task("path/to/task.json")

if success:
    # Process commands
    result = engine.process_command("go to living room")
    print(result)

Agent Interaction

from OmniSimulator.agent.agent import Agent
from OmniSimulator.core.engine import SimulationEngine

# Create simulation
engine = SimulationEngine()

# Create an agent
agent = Agent(agent_id="agent_1", name="Assistant", location_id="living_room")

# Check agent capabilities
can_grab = agent.can_grab("cup")
can_carry = agent.can_carry("table")

print(f"Agent location: {agent.location_id}")
print(f"Agent inventory: {agent.inventory}")

Action Execution

from OmniSimulator.action.action_manager import ActionManager
from OmniSimulator.core.enums import ActionType

# Initialize action manager
action_manager = ActionManager()

# Validate and execute action
validation = action_manager.validate_action(
    agent=agent,
    action_type="goto",
    target="kitchen"
)

if validation.get('valid'):
    result = action_manager.execute_action(
        agent=agent,
        action_type="goto",
        target="kitchen"
    )
    print(f"Action result: {result}")

Environment Queries

from OmniSimulator.environment.environment_manager import EnvironmentManager

# Initialize environment manager
env_manager = EnvironmentManager()

# Get room information
room = env_manager.get_room("living_room")
objects = env_manager.get_objects_in_room("living_room")

print(f"Room: {room.name}")
print(f"Objects in room: {[obj.name for obj in objects]}")

For comprehensive tutorials and examples, see Basic Simulation Examples.