flatland.envs.agent_chains module

class flatland.envs.agent_chains.ChainTestEnv(omc: MotionCheck)

Bases: object

Just for testing agent chains

addAgent(rc1, rc2, xlabel=None)

addAgentToRow(c1, c2, xlabel=None)

create_test_chain(nAgents: int, rcVel: Tuple[int] = (0, 1), liStopped: List[int] = [], xlabel=None)

create a chain of agents

nextRow()

class flatland.envs.agent_chains.MotionCheck

Bases: object

Class to find chains of agents which are “colliding” with a stopped agent. This is to allow close-packed chains of agents, ie a train of agents travelling at the same speed with no gaps between them,

addAgent(iAg, rc1, rc2, xlabel=None)

add an agent and its motion as row,col tuples of current and next position. The agent’s current position is given an “agent” attribute recording the agent index. If an agent does not want to move this round (rc1 == rc2) then a self-loop edge is created. xlabel is used for test cases to give a label (see graphviz)

block_preds(svStops, color='red')

Take a list of stopped agents, and apply a stop color to any chains/trees of agents trying to head toward those cells. Count the number of agents blocked, ignoring those which are already marked. (Otherwise it can double count swaps)

check_motion(iAgent, rcPos)

Returns tuple of boolean can the agent move, and the cell it will move into. If agent position is None, we use a dummy position of (-1, iAgent)

find_conflicts()

find_same_dest()

find groups of agents which are trying to land on the same cell. ie there is a gap of one cell between them and they are both landing on it.

find_stop_preds(svStops=None)

Find the predecessors to a list of stopped agents (ie the nodes / vertices) Returns the set of predecessors. Includes “chained” predecessors.

find_stops()

find all the stopped agents as a set of rc position nodes A stopped agent is a self-loop on a cell node.

find_stops2()

alternative method to find stopped agents, using a networkx call to find selfloop edges

find_swaps()

find all the swap conflicts where two agents are trying to exchange places. These appear as simple cycles of length 2. These agents are necessarily deadlocked (since they can’t change direction in flatland) - meaning they will now be stuck for the rest of the episode.

find_swaps2() → Set[Tuple[int, int]]

get_G_reversed()

reset_G_reversed()

flatland.envs.agent_chains.create_test_agents(omc: MotionCheck)

flatland.envs.agent_chains.create_test_agents2(omc: MotionCheck)

flatland.envs.agent_chains.main()

flatland.envs.agent_chains.render(omc: MotionCheck, horizontal=True)

flatland.envs.agent_chains.test_agent_following()