habitat跨场景室内3D仿真平台

1. Install

参考github的habitat-api及habitat-sim，注意安装其中requestment.txt文件时检查版本，大部分example运行失败都是由于功能包版本错误导致的。本机器为ubuntu18.04系统，cuda10.0版本，在anaconda环境中运行，将以上两个包都放到同一个文件夹中。通过 pip list 可以查看安装包版本如下：

absl-py (0.9.0)
astor (0.8.0)
attrs (19.3.0)
cachetools (4.1.0)
certifi (2020.4.5.1)
chardet (3.0.4)
cycler (0.10.0)
Cython (0.29.19)
decorator (4.4.2)
future (0.18.2)
gast (0.3.3)
gitdb (4.0.5)
GitPython (3.1.2)
google-auth (1.16.0)
google-auth-oauthlib (0.4.1)
grpcio (1.27.2)
gym (0.10.9)
h5py (2.8.0rc1)
habitat (0.1.5, /home/cirlab1/userdir/ybg/projects/habitat-api)
habitat-sim (0.1.5)
idna (2.9)
ifcfg (0.21)
imageio (2.2.0)
imageio-ffmpeg (0.2.0)
importlib-metadata (1.6.0)
joblib (0.15.1)
Keras-Applications (1.0.8)
Keras-Preprocessing (1.1.0)
kiwisolver (1.2.0)
llvmlite (0.31.0)
magnum (0.0.0)
Markdown (2.6.9)
matplotlib (3.2.1)
mock (2.0.0)
more-itertools (8.3.0)
moviepy (1.0.2)
numba (0.48.0)
numpy (1.16.1)
numpy-quaternion (2020.5.19.15.27.24)
oauthlib (3.1.0)
opencv-python (4.2.0.34)
packaging (20.4)
pbr (1.10.0)
Pillow (7.1.2)
pip (9.0.1)
pluggy (0.13.1)
proglog (0.1.9)
protobuf (3.11.4)
py (1.8.1)
pyasn1 (0.4.8)
pyasn1-modules (0.2.8)
pyglet (1.5.5)
pyparsing (2.4.7)
pytest (5.4.2)
python-dateutil (2.8.1)
PyYAML (5.3.1)
requests (2.23.0)
requests-oauthlib (1.3.0)
rsa (4.0)
scikit-learn (0.23.1)
scipy (1.4.1)
setuptools (47.1.1.post20200529)
six (1.15.0)
smmap (3.0.4)
tb-nightly (2.3.0a20200529)
tensorboard (1.13.1)
tensorboard-plugin-wit (1.6.0.post3)
tensorflow (1.13.1)
tensorflow-estimator (1.13.0)
termcolor (1.1.0)
threadpoolctl (2.1.0)
torch (1.2.0)
tqdm (4.46.0)
urllib3 (1.25.9)
wcwidth (0.1.9)
Werkzeug (0.12.2)
wheel (0.29.0)
yacs (0.1.7)
zipp (3.1.0)

2. Use

1. Framework

按照Github中的说明下载好相应的数据集后，可以运行各自包内的examples文件中例程，如example.py，可以验证安装是否完整。运行shortest_path_follower_example.py例程可以生成机器人的第一视角与俯视图视角视频

此demo主要用于最短路径跟随，即其中的Pointnav任务。

按照Habitat API文档介绍，对于环境的建立、动作空间选取和step后的observation提取，按如下程序：

import numpy as np
from PIL import Image
from habitat_sim.utils.common import d3_40_colors_rgbdef display_sample(rgb_obs, semantic_obs, depth_obs):rgb_img = Image.fromarray(rgb_obs, mode="RGB")semantic_img = Image.new("P", (semantic_obs.shape[1], semantic_obs.shape[0]))semantic_img.putpalette(d3_40_colors_rgb.flatten())semantic_img.putdata((semantic_obs.flatten() % 40).astype(np.uint8))semantic_img = semantic_img.convert("RGBA")depth_img = Image.fromarray((depth_obs * 255).astype(np.uint8), mode="L")arr = [rgb_img, semantic_img, depth_img]titles = ['rgb', 'semantic', 'depth']plt.figure(figsize=(12 ,8))for i, data in enumerate(arr):ax = plt.subplot(1, 3, i+1)ax.axis('off')ax.set_title(titles[i])plt.imshow(data)plt.show()config = habitat.get_config(config_paths='../configs/tasks/pointnav_mp3d.yaml')
config.defrost()
config.DATASET.DATA_PATH = '../data/datasets/pointnav/mp3d/v1/val/val.json.gz'
config.DATASET.SCENES_DIR = '../data/scene_datasets/'
config.SIMULATOR.AGENT_0.SENSORS = ['RGB_SENSOR', 'DEPTH_SENSOR', 'SEMANTIC_SENSOR']
config.SIMULATOR.SEMANTIC_SENSOR.WIDTH = 256
config.SIMULATOR.SEMANTIC_SENSOR.HEIGHT = 256
config.SIMULATOR.TURN_ANGLE = 30
config.freeze()env = habitat.Env(config=config)
env.episodes = random.sample(env.episodes, 2)max_steps = 4action_mapping = {0: 'stop',1: 'move_forward',2: 'turn left',3: 'turn right'
}for i in range(len(env.episodes)):observations = env.reset()display_sample(observations['rgb'], observations['semantic'], np.squeeze(observations['depth']))count_steps = 0while count_steps < max_steps:action = random.choice(list(action_mapping.keys()))print(action_mapping[action])observations = env.step(action)display_sample(observations['rgb'], observations['semantic'], np.squeeze(observations['depth']))count_steps += 1if env.episode_over:breakenv.close()

observation包含配置文件中设置好的数据，可以是’RGB_SENSOR’, ‘DEPTH_SENSOR’, ‘SEMANTIC_SENSOR’。通过env.step()函数，可以得到进行动作action后得到新的observation（这里只输入了observation，还可以包含reward和终止条件等。

2. Semantic information

通过环境交互，可以得到机器人第一视角下的语义图像：

该图像的不同分割区域的像素值为object对应的id值，可通过id直接获取像素表示的物体类别、所属房间、Bounding box等信息。本部分内容Habitat官方文档并未涉及，在此进行展开叙述：

# 获取该仿真场景的语义信息
scene = env.habitat_env.sim.semantic_annotations()
# 环境交互 获取observation
observations, reward, done, info = env.step(best_action)
# 获取语义分割图像 像素值为对应object的id
semantic_obs = observations["semantic"]
# 使用flatten()函数将图像二维矩阵转化为一维数组，然后使用np.bincount计算某个值在整个数组中的次数
counts = np.bincount(semantic_obs.flatten())
# 计算总次数
total_count = np.sum(counts)
image_count += 1
for object_i, count in enumerate(counts):sem_obj = scene.objects[object_i]cat = sem_obj.category.name()pixel_ratio = count / total_countif pixel_ratio > 0.01 :print(f"obj_id:{sem_obj.id},category:{cat}, pixel_ratio:{pixel_ratio}")

在获取的具有语义标注的scene数据中，包含三次数据结构，分别是levels，regions和objects，每层数据结构都有自己的属性，可调用性如下：

r"""Returns:SemanticScene which is a three level hierarchy of semanticannotations for the current scene. Specifically this methodreturns a SemanticScene which contains a list of SemanticLevel'swhere each SemanticLevel contains a list of SemanticRegion's whereeach SemanticRegion contains a list of SemanticObject's.SemanticScene has attributes: aabb(axis-aligned bounding box) whichhas attributes aabb.center and aabb.sizes which are 3d vectors,categories, levels, objects, regions.SemanticLevel has attributes: id, aabb, objects and regions.SemanticRegion has attributes: id, level, aabb, category (to getname of category use category.name()) and objects.SemanticObject has attributes: id, region, aabb, obb (orientedbounding box) and category.SemanticScene contains List[SemanticLevels]SemanticLevel contains List[SemanticRegion]SemanticRegion contains List[SemanticObject]Example to loop through in a hierarchical fashion:for level in semantic_scene.levels:for region in level.regions:for obj in region.objects:
"""
scene = sim.semantic_scenefor obj in scene.objects:obj.aabbobj.aabb.sizesobj.aabb.centerobj.idobj.obb.rotationobj.category.name()obj.category.index()for region in scene.regions:region.idregion.category.name()region.category.index()for level in scene.levels:level.id

根据此数据结构，以上代码可以输出语义图像对应的id，类别，并且根据在一帧图像中出现的比重，滤掉噪点数据。采集的数据如下：

obj_id:1_8_206,category:door, pixel_ratio:0.23677083333333335
obj_id:1_8_211,category:wall, pixel_ratio:0.22867513020833333
obj_id:1_8_215,category:misc, pixel_ratio:0.04415364583333333
obj_id:1_8_221,category:picture, pixel_ratio:0.010263671875
obj_id:1_9_235,category:wall, pixel_ratio:0.035143229166666665
obj_id:1_9_239,category:cabinet, pixel_ratio:0.01310546875
obj_id:1_9_260,category:floor, pixel_ratio:0.011048177083333333
obj_id:1_20_463,category:wall, pixel_ratio:0.100205078125
obj_id:1_20_469,category:lighting, pixel_ratio:0.012141927083333334
obj_id:1_20_471,category:sofa, pixel_ratio:0.095908203125
obj_id:1_20_490,category:floor, pixel_ratio:0.082001953125
obj_id:1_21_498,category:wall, pixel_ratio:0.012578125
obj_id:1_22_505,category:wall, pixel_ratio:0.066162109375

以上是一帧语义图像所包含的语义信息。obj_id中，第一位数据为level层，表示建筑物的id；第二位数据为region层，表示具体场景的id，如bedroom等；第三位数据为object层，表示object的id（即为语义图像的像素值）。根据obj_id即可判断出该语义物体的类别、房间和建筑物的所属关系等。

3. RL framework

下面介绍该框架使用RL的方法，分别从Agent、Environment、Task进行展开。

1. Agent

Agent的动作集包含：

POSSIBLE_ACTIONS: ["STOP", "MOVE_FORWARD", "TURN_LEFT", "TURN_RIGHT", "LOOK_UP", "LOOK_DOWN"]

2. Environment

Environment在habitat代码框架中的baseline定义于habitat_baseline/common/environment.py中，通过registry注册名字为NavRLEnv，可以在配置文件中直接设置整个环境类NavRLEnv继承了/habitat/core/env.py中的RLEnv类，而RLEnv类则是继承自gym环境中的Env，专门用于RL任务的封装。

对于环境类中的各种操作，分别对父类中的函数进行继承或重写，下面介绍RL框架中需要用到的函数及其定义的位置：

1. reset函数：重置环境，返回重置后的观测

def reset(self):self._previous_action = Noneobservations = super().reset()self._previous_measure = self._env.get_metrics()[self._reward_measure_name]return observations

该函数在子类NavRLEnv中对父类/habitat/core/env.py中的RLEnv类中的step进行了重写。

2. step函数：接收action值，返回在仿真环境中的观测、奖励、是否完成任务等信息

def step(self, *args, **kwargs) -> Tuple[Observations, Any, bool, dict]:r"""Perform an action in the environment.:return: :py:`(observations, reward, done, info)`"""observations = self._env.step(*args, **kwargs)reward = self.get_reward(observations)done = self.get_done(observations)info = self.get_info(observations)return observations, reward, done, info

该函数定义于/habitat/core/env.py中的RLEnv类中，在后面的子类NavRLEnv中直接继承。

3. get_reward函数：获取当前观测的奖励值，输入当前观测observation，输出奖励值reward

其中的奖励函数按照论文中的描述：

Let $r_{t}$ denote the reward at timestep $t, d_{t}$ be the geodesic distance to goal at timestep $t, s$ a success reward and $\lambda$ a time penalty (to encourage efficiency). All models were trained with the following reward function:
$r_{t}=\left\{\begin{array}{ll} s+d_{t-1}-d_{t}+\lambda & \text { if goal is reached } \\ d_{t-1}-d_{t}+\lambda & \text { otherwise } \end{array}\right.$
在habitat中代码形式如下：

def get_reward(self, observations):reward = self._rl_config.SLACK_REWARDcurrent_measure = self._env.get_metrics()[self._reward_measure_name]reward += self._previous_measure - current_measureself._previous_measure = current_measureif self._episode_success():reward += self._rl_config.SUCCESS_REWARDreturn reward

该函数在子类NavRLEnv中对父类/habitat/core/env.py中的RLEnv类中的step进行了重写。

4. get_done函数：判断是否完成导航任务，输入为当前观测observation，输出为bool值

论文中关于完成任务的标准描述如下：

A navigation episode is considered successful if and only if the agent issues a stop action within 0.2m of the target coordinates, as measured by a geodesic distance along the shortest path from the agent’s position to the goal position. If the agent takes 500 actions without the above condition being met the episode ends and is considered unsuccessful.

即只有输出动作stop，并且在最短路径上距target距离小于0.2m即可判断成功，若运行超过500步仍未完成任务，则视为任务失败。

在habitat中代码形式如下：

def get_done(self, observations):done = Falseif self._env.episode_over or self._episode_success():done = Truereturn done

该函数在子类NavRLEnv中对父类/habitat/core/env.py中的RLEnv类中的step进行了重写。

5. _episode_success函数：判断任务是否完成，在上面的reward函数和done函数中都用到了该函数。

def _episode_success(self):return self._env.get_metrics()[self._success_measure_name]

该函数在NavRLEnv类中定义，其主要调用了self._env.get_metrics()函数用来获得结果。其中的env来自父类RLEnv中 init _ 的定义：

self._env = Env(config, dataset)

而Env类则是habitat最基础的类，同样定义于env.py文件中。

def get_metrics(self) -> Metrics:return self._task.measurements.get_metrics()

对于_task的定义同样在Env类中：

self._task = make_task(self._config.TASK.TYPE,   # TYPE: ObjectNav-v1config=self._config.TASK,sim=self._sim,dataset=self._dataset,)

继续往下找，make_task的定义在/habitat/task/registration.py中：

def make_task(id_task, **kwargs):logger.info("Initializing task {}".format(id_task))_task = registry.get_task(id_task)assert _task is not None, "Could not find task with name {}".format(id_task)return _task(**kwargs)

由此可知，在env创建初始化时，即完成了任务的注册，注册名为ObjectNav-v1，并最后返回_task的实例化。根据registry文件中task的注册代码：

@classmethoddef register_task(cls, to_register=None, *, name: Optional[str] = None):r"""Register a task to registry with key :p:`name`:param name: Key with which the task will be registered.If :py:`None` will use the name of the class.. code:: pyfrom habitat.core.registry import registryfrom habitat.core.embodied_task import EmbodiedTask@registry.register_taskclass MyTask(EmbodiedTask):pass# or@registry.register_task(name="MyTaskName")class MyTask(EmbodiedTask):pass"""from habitat.core.embodied_task import EmbodiedTaskreturn cls._register_impl("task", to_register, name, assert_type=EmbodiedTask)

可知，该注册的task，注册名为ObjectNav-v1，以habitat.core.embodied_task中的EmbodiedTask类为父类，找到位于habitat/task/nav/object_nav_task.py中的ObjectNavigationTask子类：

@registry.register_task(name="ObjectNav-v1")
class ObjectNavigationTask(NavigationTask):r"""An Object Navigation Task class for a task specific methods.Used to explicitly state a type of the task in config."""pass

由于该类没有进行操作，继续查找其父类，位于habitat/task/nav/nav.py中的NavigationTask类：

@registry.register_task(name="Nav-v0")
class NavigationTask(EmbodiedTask):def __init__(self, config: Config, sim: Simulator, dataset: Optional[Dataset] = None) -> None:super().__init__(config=config, sim=sim, dataset=dataset)def overwrite_sim_config(self, sim_config: Any, episode: Type[Episode]) -> Any:return merge_sim_episode_config(sim_config, episode)def _check_episode_is_active(self, *args: Any, **kwargs: Any) -> bool:return not getattr(self, "is_stop_called", False)

该类也没有实际操作，因此最终的函数定义到位于habitat/core/embodied_task.py基类EmbodiedTask.

即make_task中后三项输入正好对应于EmbodiedTask类实例化参数：

def __init__(self, config: Config, sim: Simulator, dataset: Optional[Dataset] = None) -> None:self._config = configself._sim = simself._dataset = dataset

由此可知，对于任务是否完成的度量来自EmbodiedTask类，下面针对task继续挖掘。

3. Task

关于task，论文中描述如下：

-Task: this class extends the simulator's observations class and action space with taskspecific ones. The criteria of episode termination and measures of success are provided by the Task. For example, in goal-driven navigation, Task provides the goal and evaluation metric.

可以看出，task大类主要就是对episode进行审判，是否任务终止和任务的成功率，并提供了到goal的距离。

继续从上节开始，实例化EmbodiedTask类时传入config中的TASK配置参数，即self._task.measurements.get_metrics()中的 _task实例化EmbodiedTask类，measurements又实例化了Measurements类：

self.measurements = Measurements(self._init_entities(entity_names=config.MEASUREMENTS,register_func=registry.get_measure,entities_config=config,).values()
)

需要注意，在实例化Measurements时，def _init_entitie函数最后返回一个dict，初始化参数为字典self._init_entities的values值以返回该字典中的所有值。这里输入参数config.MEASUREMENTS即为需测量的参数数组[‘DISTANCE_TO_GOAL’, ‘SUCCESS’, ‘SPL’, ‘SOFT_SPL’]，registry.get_measure为注册的测量信息，符合的函数在nav.py中，有：Success，SPL，SoftSPL，Collisions，TopDownMap这几个Measure子类。

_init_entities函数返回一个字典，分别是MEASUREMENTS需测量的参数，和对应参数的注册函数register_func：

def _init_entities(self, entity_names, register_func, entities_config=None
) -> OrderedDict:if entities_config is None:entities_config = self._configentities = OrderedDict()for entity_name in entity_names:entity_cfg = getattr(entities_config, entity_name)entity_type = register_func(entity_cfg.TYPE)assert (entity_type is not None), f"invalid {entity_name} type {entity_cfg.TYPE}"entities[entity_name] = entity_type(sim=self._sim,config=entity_cfg,dataset=self._dataset,task=self,)return entities

在Measurements类中，找到了get_metrics()函数：

def get_metrics(self) -> Metrics:r"""Collects measurement from all :ref:`Measure`\ s and returns itpackaged inside :ref:`Metrics`."""return Metrics(self.measures)

该函数的输入即为一个dict，索引为：success、spl、softspl等测量参数字符串，值为对应的注册测量函数，函数类为Measure：

class Metrics(dict):r"""Dictionary containing measurements."""def __init__(self, measures: Dict[str, Measure]) -> None:"""Constructor:param measures: list of :ref:`Measure` whose metrics are fetched andpackaged."""data = [(uuid, measure.get_metric()) for uuid, measure in measures.items()]super().__init__(data)

即self._task.measurements.get_metrics()最终返回的就是这个Metrics类，是一个包含measurements的dict，包含id和measure数据。通过get_metrics()[" "]即可查询对应的评估参数。其中，measures: Dict[str, Measure]由一个str类型的id和Measure类组成，data包含的uuid为str，measure为Measure类的数据，即Measure类的子类，在这里即可揭开如何评估任务的面纱！

1. Success

首先是成功判断，即Measure的子类Success：

@registry.register_measure
class Success(Measure):r"""Whether or not the agent succeeded at its taskThis measure depends on DistanceToGoal measure."""cls_uuid: str = "success"def __init__(self, sim: Simulator, config: Config, *args: Any, **kwargs: Any):self._sim = simself._config = configsuper().__init__()def _get_uuid(self, *args: Any, **kwargs: Any) -> str:return self.cls_uuiddef reset_metric(self, episode, task, *args: Any, **kwargs: Any):task.measurements.check_measure_dependencies(self.uuid, [DistanceToGoal.cls_uuid])self.update_metric(episode=episode, task=task, *args, **kwargs)def update_metric(self, episode, task: EmbodiedTask, *args: Any, **kwargs: Any):distance_to_target = task.measurements.measures[DistanceToGoal.cls_uuid].get_metric()if (hasattr(task, "is_stop_called")and task.is_stop_calledand distance_to_target < self._config.SUCCESS_DISTANCE):self._metric = 1.0else:self._metric = 0.0

待更