OmniSimulator

Embodied Simulator - 文本具身任务模拟器

这个包提供了用于模拟智能体在虚拟环境中执行文本具身任务的工具。

使用方式：: from simulator import SimulationEngine, ActionHandler, SimulationLogger from simulator import ActionType, ActionStatus, ObjectType

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

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

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

初始化模拟引擎

Parameters:

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

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

获取智能体信息

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

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

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

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

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

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

获取任务验证状态

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

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

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

初始化模拟器 :param scene_file: 场景文件路径 :param agent_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

initialize_with_task(task_file: str) → bool[source]

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

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

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

restart_visualization() → bool[source]

重启可视化系统

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

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

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

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

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

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

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

Parameters:: new_abilities – 新的能力列表

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

验证当前环境是否合法

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

class OmniSimulator.WorldState[source]

世界状态类 - 维护整个模拟世界的状态

__init__()[source]

add_agent(agent_id: str, agent_data: Dict) → None[source]: 添加智能体到世界状态

describe_agent_natural_language(agent_id: str, agent: Dict = None) → str[source]: 转发到EnvironmentGraph的describe_agent_natural_language方法

describe_environment_natural_language(agents: Dict[str, Dict] | None = None, sim_config: Dict[str, Any] | None = None) → str[source]: 转发到EnvironmentGraph的describe_environment_natural_language方法

describe_room_natural_language(room_id: str, agents: Dict[str, Dict] | None = None, sim_config: Dict[str, Any] | None = None) → str[source]: 转发到EnvironmentGraph的describe_room_natural_language方法

get_agent(agent_id: str) → Dict | None[source]: 获取智能体数据

get_object(object_id: str) → Dict | None[source]: 获取物体数据

is_object_accessible_to_agent(object_id: str, agent_id: str) → Tuple[bool, str][source]: 检查物体是否对智能体可访问

update_agent(agent_id: str, update_data: Dict) → None[source]: 更新智能体状态

class OmniSimulator.ActionType(value)[source]

动作类型枚举

GOTO = 1

GRAB = 2

PLACE = 3

LOOK = 4

EXPLORE = 5

CORP_GRAB = 6

CORP_GOTO = 7

CORP_PLACE = 8

ATTRIBUTE = 9

class OmniSimulator.ActionStatus(value)[source]

动作执行状态枚举

SUCCESS = 1

FAILURE = 2

INVALID = 3

PARTIAL = 4

WAITING = 5

class OmniSimulator.ObjectType(value)[source]

物体类型枚举

STATIC = 1

INTERACTABLE = 2

GRABBABLE = 3

FURNITURE = 4

ITEM = 5

class OmniSimulator.ActionHandler(world_state, env_manager, agent_manager, scene_abilities=None, config=None)[source]

动作处理器 - 负责处理智能体的动作请求

此类是模拟器的主要对外接口，用于处理智能体的动作请求。它封装了ActionManager的内部复杂性，提供简单直观的API。

使用示例: `python action_handler = ActionHandler(world_state, env_manager, agent_manager) status, message, result = action_handler.process_command('agent_1', 'GRAB cup_1') `

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

初始化动作处理器

Parameters:

world_state – 世界状态对象
env_manager – 环境管理器
agent_manager – 智能体管理器
scene_abilities – 场景支持的能力列表，用于注册不需要工具的属性动作
config – 配置字典，包含验证相关配置

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

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

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

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

获取待处理的合作动作

Returns:: 动作ID到动作数据的映射
Return type:: Dict[str, Any]

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

手动获取任务验证状态（用于手动查询）

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

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

处理智能体的命令

Parameters:

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

Returns:

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

Return type:

Tuple[ActionStatus, str, Optional[Dict]]

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

注册单个动作类

Parameters:

action_name – 动作名称
action_class – 动作类

Returns:

支持链式调用

Return type:

self

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

批量注册多个动作类

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

register_attribute_actions()[source]

加载属性动作配置（仅为保持向后兼容）

Returns:: 支持链式调用
Return type:: self

set_task_verifier(task_data: Dict[str, Any])[source]

设置任务验证器

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

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

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

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

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

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

初始化动作管理器

Parameters:

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

agent_action_classes = {}

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

执行动作

Parameters:: action – 动作对象
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

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]]

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]]

classmethod register_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

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

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

Parameters:

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

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

验证动作是否有效

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

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

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

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

__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

command_pattern: ClassVar[str] = None

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]]

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

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

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

验证动作是否可执行

Parameters:

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

Returns:

(是否有效, 原因消息)

Return type:

Tuple[bool, str]

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

GOTO动作 - 移动到指定位置

action_type: ClassVar[ActionType] = 1

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

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

GRAB动作 - 抓取物体

action_type: ClassVar[ActionType] = 2

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

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

PLACE动作 - 放置物体

action_type: ClassVar[ActionType] = 3

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

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

LOOK动作 - 观察环境或物体

action_type: ClassVar[ActionType] = 4

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

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

EXPLORE动作 - 探索当前房间

action_type: ClassVar[ActionType] = 5

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

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

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

CSV格式: action_name,attribute,value,requires_tool

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

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

action_configs = {}

action_type: ClassVar[ActionType] = 9

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

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

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

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

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

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

no_tool_actions = {}

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 – 任务中指定的能力列表

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

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

__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等

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

从物体获取能力

Parameters:

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

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]

can_grab() → bool[source]

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

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

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]

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

从字典创建智能体对象

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

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]

has_ability(ability: str) → bool[source]

检查是否拥有特定能力

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

move_to(new_location_id: str) → None[source]

移动到新位置

Parameters:: new_location_id – 新位置ID

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

移除来自特定物体的能力

Parameters:

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

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

将智能体转换为字典表示

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

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

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

Parameters:

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

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

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

__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

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

将物体添加到智能体库存

Parameters:

agent_id – 智能体ID
object_id – 物体ID

Returns:

添加是否成功

Return type:

bool

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

获取智能体实例

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

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]

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

移动智能体到新位置

Parameters:

agent_id – 智能体ID
new_location_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

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

更新智能体数据

Parameters:

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

Returns:

更新是否成功

Return type:

bool

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

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

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

初始化环境管理器

Parameters:

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

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

add_room(room: Room) → bool[source]

添加房间到环境

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

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

连接两个房间

Parameters:

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

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]

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]

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]

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

获取物体数据

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

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

获取物体位置

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

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

获取物体所在的房间ID

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

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_room_by_id(room_id: str) → Dict[str, Any] | None[source]

获取房间数据

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

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]

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]

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

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

parse_location_id(location_id: str)[source]

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

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

更新物体状态

Parameters:

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

Returns:

更新是否成功

Return type:

bool

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

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

__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

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

is_connected_to(room_id: str) → bool[source]

检查是否连接到特定房间

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

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

class OmniSimulator.BaseObject(obj_id: str, name: str, object_type: ObjectType, states: Dict[str, Any] | None = None, properties: Dict[str, Any] | None = None)[source]

基础物体类

__init__(obj_id: str, name: str, object_type: ObjectType, states: Dict[str, Any] | None = None, properties: Dict[str, Any] | None = None)[source]

初始化基础物体

Parameters:

obj_id – 物体唯一ID
name – 物体名称
object_type – 物体类型枚举值
states – 物体状态字典，如 {‘is_open’: False, ‘is_on’: True}
properties – 物体属性字典，如 {‘color’: ‘red’, ‘size’: ‘small’}

classmethod from_dict(data: Dict[str, Any]) → BaseObject[source]: 从字典创建物体对象

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

class OmniSimulator.StaticObject(obj_id: str, name: str, states: Dict[str, Any] | None = None, properties: Dict[str, Any] | None = None)[source]

静态物体类

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

初始化静态物体

Parameters:

obj_id – 物体唯一ID
name – 物体名称
states – 物体状态字典
properties – 物体属性字典

class OmniSimulator.InteractableObject(obj_id: str, name: str, states: Dict[str, Any] | None = None, properties: Dict[str, Any] | None = None)[source]

可交互物体类

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

初始化可交互物体

Parameters:

obj_id – 物体唯一ID
name – 物体名称
states – 物体状态字典，如 {‘is_open’: False, ‘is_on’: True}
properties – 物体属性字典

class OmniSimulator.GrabbableObject(obj_id: str, name: str, states: Dict[str, Any] | None = None, properties: Dict[str, Any] | None = None)[source]

可抓取物体类

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

初始化可抓取物体

Parameters:

obj_id – 物体唯一ID
name – 物体名称
states – 物体状态字典
properties – 物体属性字典，如 {‘weight’: 0.5, ‘size’: ‘small’}

class OmniSimulator.SceneParser[source]

场景解析器 - 用于解析场景和（已废弃：智能体只能通过simulator_config.yaml加载）

__init__()[source]: 初始化场景解析器

parse_file(file_path: str) → Dict[str, Any] | None[source]

根据文件扩展名选择适当的解析方法

Parameters:: file_path – 文件路径
Returns:: 解析后的数据字典，解析失败则返回None

parse_scene_file(file_path: str) → Dict[str, Any] | None[source]

解析场景配置文件

Parameters:: file_path – 场景配置文件路径
Returns:: 解析后的场景数据字典，解析失败则返回None

class OmniSimulator.SceneValidator[source]

场景验证器 - 用于检查场景数据的合法性

static validate_agent_config(agent_config: Dict[str, Any]) → Tuple[bool, List[str]][source]

验证智能体配置

Parameters:: agent_config – 智能体配置字典
Returns:: (是否合法, 错误信息列表)
Return type:: Tuple[bool, List[str]]

static validate_agent_file(file_path: str) → Tuple[bool, List[str]][source]

验证智能体配置文件

Parameters:: file_path – 智能体配置文件路径
Returns:: (是否合法, 错误信息列表)
Return type:: Tuple[bool, List[str]]

static validate_container_relationships(scene_data: Dict[str, Any]) → List[str][source]: 验证场景中的容器关系合法性

static validate_scene(scene_data: Dict[str, Any]) → Tuple[bool, List[str]][source]

验证场景数据的合法性

Parameters:: scene_data – 场景数据字典
Returns:: (是否合法, 错误信息列表)
Return type:: Tuple[bool, List[str]]

static validate_scene_file(file_path: str) → Tuple[bool, List[str]][source]

验证场景文件

Parameters:: file_path – 场景文件路径
Returns:: (是否合法, 错误信息列表)
Return type:: Tuple[bool, List[str]]

class OmniSimulator.SimulationLogger(log_level: str = 'INFO', log_file: str | None = None, log_to_console: bool = True)[source]

模拟器日志工具 - 负责记录模拟器操作日志

__init__(log_level: str = 'INFO', log_file: str | None = None, log_to_console: bool = True)[source]

初始化日志工具

Parameters:

log_level – 日志级别，可选值为DEBUG, INFO, WARNING, ERROR, CRITICAL
log_file – 日志文件路径，如果为None则不保存到文件
log_to_console – 是否输出到控制台

critical(message: str) → None[source]: 记录CRITICAL级别日志

debug(message: str) → None[source]: 记录DEBUG级别日志

end_session() → None[source]: 记录会话结束

error(message: str) → None[source]: 记录ERROR级别日志

info(message: str) → None[source]: 记录INFO级别日志

log_action(agent_id: str, action: str, status: str, message: str) → None[source]

记录动作日志

Parameters:

agent_id – 智能体ID
action – 动作命令
status – 执行状态
message – 反馈消息

log_error(error_type: str, message: str) → None[source]

记录错误日志

Parameters:

error_type – 错误类型
message – 错误消息

log_state_change(entity_id: str, property_name: str, old_value: Any, new_value: Any) → None[source]

记录状态变化日志

Parameters:

entity_id – 实体ID (物体或智能体)
property_name – 变化的属性名
old_value – 旧值
new_value – 新值

start_session(config: Dict[str, Any]) → None[source]

记录会话开始

Parameters:: config – 配置信息

warning(message: str) → None[source]: 记录WARNING级别日志

class OmniSimulator.FeedbackGenerator(verbose: bool = False)[source]

反馈生成器 - 负责生成用户友好的反馈信息

__init__(verbose: bool = False)[source]

初始化反馈生成器

Parameters:: verbose – 是否生成详细反馈

format_feedback(status: ActionStatus, message: str, result_data: Dict[str, Any] | None = None) → str[source]

格式化反馈消息

Parameters:

status – 动作执行状态
message – 原始反馈消息
result_data – 额外的结果数据

Returns:

格式化后的反馈消息

Return type:

str

generate_observation(agent_id: str, location_data: Dict[str, Any], objects_data: Dict[str, Any]) → str[source]

生成环境观察描述

Parameters:

agent_id – 智能体ID
location_data – 位置数据
objects_data – 物体数据

Returns:

观察描述

Return type:

str

class OmniSimulator.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 – 验证配置

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

获取任务完成情况摘要

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

get_current_completion_status() → 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_all_tasks() → List[TaskVerificationResult][source]

验证所有任务

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

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

验证单个子任务

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

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

验证特定类别的任务

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

class OmniSimulator.VisualizationManager(world_state, agent_manager, env_manager, config_path: str | None = None, engine=None)[source]

可视化管理器 - 统一管理可视化系统

__init__(world_state, agent_manager, env_manager, config_path: str | None = None, engine=None)[source]

初始化可视化管理器

Parameters:

world_state – 世界状态对象
agent_manager – 智能体管理器
env_manager – 环境管理器
config_path – 配置文件路径
engine – 模拟引擎实例

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

获取特定房间的数据

Parameters:: room_id – 房间ID
Returns:: 房间数据，如果未启用则返回None
Return type:: Dict

get_server_url() → str | None[source]: 获取Web服务器URL

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

获取可视化系统状态

Returns:: 状态信息
Return type:: Dict

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

获取当前可视化数据

Returns:: 可视化数据，如果未启用则返回None
Return type:: Dict

initialize() → bool[source]

初始化可视化系统

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

is_visualization_enabled() → bool[source]: 检查可视化是否启用

is_visualization_running() → bool[source]: 检查可视化是否在运行

start() → bool[source]

启动可视化系统

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

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

update_config(new_config: Dict[str, Any])[source]

更新配置

Parameters:: new_config – 新的配置字典

class OmniSimulator.VisualizationWebServer(data_provider, config: Dict | None = None)[source]

可视化Web服务器

__init__(data_provider, config: Dict | None = None)[source]

初始化Web服务器

Parameters:

data_provider – 数据提供器
config – 服务器配置

get_server_url() → str[source]: 获取服务器URL

is_running() → bool[source]: 检查服务器是否在运行

start()[source]: 启动Web服务器

stop()[source]: 停止Web服务器