基于多agent的Get_state()函数问题

Question

@Ashkan我看到了你对这个问题的答案(state() return based in multi-agent based on agent-id?)

您给出了一些示例代码: def get_state(self)：

    agent_state_dict = {}
    i = 0
    for intersection, edges in self.scenario.get_node_mapping():
        i = i + 1
        agent_id = self.agent_name_prefix + str(i) # self.agent_name_prefix is defined as string "intersection"

        speeds = []
        dist_to_intersec = []
        traffic_light_states = []

        ..... code .....

        # construct the state (observation) for each agent
        observation = np.array(
            np.concatenate([
                speeds, dist_to_intersec, traffic_light_states  

        # each intersection is an agent, so we will make a dictionary that maps form "self.agent_name_prefix+'i'" to the state of that agent.
        agent_state_dict.update({agent_id: observation})

    return agent_state_dict

我对你的代码有疑问：

1. 在“for”循环中，您只使用过一次交集和边，而没有其他的交集和边的用法，在这里，交点和边的函数是什么？
2. 在dist_to_intesec[]上，基于green_wave_env.py，它将返回所有车辆到所有交叉口的距离，而不是将车辆的距离返回到一个特殊的/个别的交叉口，我不太理解您在这里的dist_to_intersec[]，您能解释一下吗？
3. 如何检查get_state()的数据，例如，我想获得dist_to_intersec的数据。

在我的项目:基于网格网络，我想得到多少车辆是在一个交叉口的圆(例如，半径为100米，交叉口是中心点)的地平线时间。所以你的回答会对我有很大帮助。@Ashkan

Answer 1

在那篇文章中，为了更好的理解，我刚刚发布了一个代码的骨架，而完整的代码并没有显示在那里。

以下是你的问题的答案：

1. intersection稍后在self.k.traffic_light.get_state(intersection)中用于获取交叉口中交通灯的状态。edges被用作observed_vehicle_ids = self.k_closest_to_intersection_edge(edges, self.num_closest_vehicles_onbound)函数的输入，以获取指定边缘上的cars。
2. dist_to_intersec是一个变量，它将简单地存储存储在observed_vehicle_ids中的观测车辆的距离(并非所有车辆到所有交叉口的距离)。

for veh_id in observed_vehicle_ids:
    if veh_id == 0:
        dist_to_intersec.append(-1)
        speeds.append(-1)
    else:
        dist_to_intersec.append(
            (self.k.scenario.edge_length(
                self.k.vehicle.get_edge(veh_id))
                - self.k.vehicle.get_position(veh_id)) / max_dist
        )
        speeds.append(
            self.k.vehicle.get_speed(veh_id) / max_speed
        )

1. 你可以打印状态。

要在交叉口的每个边缘获取固定数量的车辆，您可以尝试如下：

def k_closest_to_intersection_edge(self, edges, k):
    """
    Return the veh_id of the 4*k closest vehicles to an intersection for
    each edge (k closest vehicles on each edge). 
    """
    if k < 0:
        raise IndexError("k must be greater than 0")
    ids = []

    def sort_lambda(veh_id):
        return self.get_distance_to_intersection(veh_id)

    for edge in edges:
        vehicles = self.k.vehicle.get_ids_by_edge(edge)
        veh_ids_per_bound = sorted(
            vehicles,
            key=sort_lambda
        )
        if len(veh_ids_per_bound) >= k: # we have more than k vehicles, and we need to cut
            ids += veh_ids_per_bound[:k]
        else: # we have less than k vehicles, and we need to pad
            padding = k - len(veh_ids_per_bound)
            ids += (veh_ids_per_bound + [0]*padding)

    return ids