Example simulation using NXRouter¶
This is the simpler of the two tutorials for it does not require a running BRouter server.
Jupyter interactive tutorial¶
You can download the following notebook from https://gitlab.com/rgarcia-herrera/road-agent/blob/master/doc/jupyter_tutorial_NXRouter.ipynb to run a local copy and do some experiments.
Try changing speeds for modes, initial population sizes. Try other maps. Enjoy!
A game of tag¶
To exemplify the use of the Road Agent framework we shall write a game of tag simulation.
Agents are bike riders. Tagged bikes -which are faster- seek untagged riders. If they come close enough to untagged bikes, they tag them! To increase their chances of catching untagged bikes, they randomly circle nodes with the highest betweenness centrality in the street network.
Untagged riders must cross town from point A to point B and back again continuously. As an evasive strategy untagged bikes route through points with the lowest betweeness centrality in the street network.
import osmnx as ox
%matplotlib inline
import matplotlib.pyplot as plt
ox.config(log_file=True, log_console=True, use_cache=True)
from LatLon import LatLon
# grab a square somewhere in Mexico City
G = ox.graph_from_point((19.3838,-99.1758), distance=1200)
fig, ax = ox.plot_graph(G, node_size=0)
# you could also use this method and run a simulation on your home town. Try it!
#G = ox.graph_from_place('Tzintzuntzan, Michoacan, Mexico', network_type='bike')
Set bases A and B¶
Untagged bikes will run among these, to and fro.
# find westernmost node to make it base A
x_min = min([G.nodes[osmid]['x'] for osmid in G.nodes])
base_A = [G.nodes[osmid] for osmid in G.nodes if G.nodes[osmid]['x'] == x_min][0]
base_A_point = LatLon(base_A['y'], base_A['x'])
base_A
# find easternmost node to make it base B
x_max = max([G.nodes[osmid]['x'] for osmid in G.nodes])
base_B = [G.nodes[osmid] for osmid in G.nodes if G.nodes[osmid]['x'] == x_max][0]
base_B_point = LatLon(base_B['y'], base_B['x'])
base_B
fig, ax = ox.plot_graph(G, node_size=0, show=False, close=False)
ax.add_artist(plt.Circle((base_A['x'], base_A['y']), 0.001, color='red', alpha=0.9))
ax.add_artist(plt.Circle((base_B['x'], base_B['y']), 0.001, color='blue', alpha=0.9))
plt.show()
Low betweenness centrality nodes¶
Crate a list of low betweenness centrality nodes. We'll grab items from here to include in the untagged routes, according to our evasion strategy.
lower_betweeness_threshold = 0.04
import networkx as nx
from LatLon import LatLon
import random
import numpy as np
low_btw = [LatLon(G.node[osmid]['y'], G.node[osmid]['x'])
for osmid, btw in nx.betweenness_centrality(G).iteritems()
if btw > 0 and btw <= lower_betweeness_threshold]
print len(low_btw), "low betweenness nodes, out of a total of ", len(G.node)
# a random node from this dict
random.choice(low_btw)
High betweeness centrality nodes¶
Tagged agents will roam the most central nodes. Victims are shure to come by!
high_betweeness_threshold = 0.1
high_btw = [LatLon(G.node[osmid]['y'], G.node[osmid]['x'])
for osmid, btw in nx.betweenness_centrality(G).iteritems()
if btw >= high_betweeness_threshold]
# there's just a few of them
high_btw
The Bike class¶
Now we extend the Agent class to create our bikes. We'll write methods for both modes of play: tagged and untagged.
from road_agent import Agent
class Bike(Agent):
def dest_high_btw_node(self):
"""
tagged agents will ride to high betweenness nodes
to increase their chances of catching untagged agents
"""
self.set_destination(random.choice(high_btw))
self.update_route()
def set_mode(self, mode):
if mode == 'tagged':
self.mode = 't'
# tagged bikes go faster
self.speed = random.uniform(4, 6) # speed given in m/s
# just got tagged? select random destination
self.dest_high_btw_node()
else:
self.mode = 'u'
self.speed = random.uniform(3, 4)
# choose a route through low betweeness node
b.update_route([random.choice(low_btw), ])
def tag_nearby_agents(self, bikes):
"""
seek bikes close to me, tag them!
"""
for b in bikes:
if self.distance_to(b.point()) < self.speed and b.mode == 'u':
print "at t=%s agent %s tagged %s!" % (t, id(self), id(b))
b.set_mode('tagged')
break # tag just one at a time!
def tagged_step(self):
# Here we use the global variable N which contains the whole bike population
# 'cause this is a tutorial and it is useful to keep it simple.
# But you might use a database with useful queries and such
# by mixing in the Agent class with an ORM
self.tag_nearby_agents(N)
# choose a destination
if self.got_there():
self.dest_high_btw_node()
else:
self.step()
def untagged_step(self):
if self.got_there():
# got to a base? turn around and head back to the one you came from!
if self.destination() == base_A_point:
self.set_destination(base_B_point)
elif self.destination() == base_B_point:
self.set_destination(base_A_point)
# but go through a low betweenness centrality node, to try and evade taggers
self.update_route([random.choice(low_btw), ])
else:
self.step()
def play_tag(self):
if self.mode == 't':
self.tagged_step()
elif self.mode == 'u':
self.untagged_step()
Initialize NXRouter¶
The Road Agent framework is distributed with two router alternatives:
- NXRouter, wich uses shortest paths computed with the NetworkX library.
- BRouter wich uses a BRouter server to query routes.
NXRouter is simpler but easier to run. Some directed graphs will fail to find paths between nodes. BRouter is very thorough but it requires the installation and running of the BRouter server which is a java program and maybe a long download.
Here we shall use the NXRouter. To keep it simple, we convert the directed street network to an undirected one, which makes all streets run both ways, but makes it easy to route from any point in the graph to any other point.
from road_agent.router import NXRouter
undirected_G = nx.to_undirected(G)
# NXRouter is initialized with a graph
router = NXRouter(undirected_G)
Simulation initialization¶
number_of_bikes = 20
simulation_steps = 2400
# log population sizes
tagged_pop = np.zeros(simulation_steps)
untagged_pop = np.zeros(simulation_steps)
# create a bike population with a list comprehension
N = [Bike(point=base_A_point, dest=base_B_point, # all bikes start at base A, heading for B
router=router)
for n in range(number_of_bikes)]
for b in N:
# set them all in the untagged mode
b.set_mode('untagged')
# let each agent keep a log of its trail
b.trail = []
# create one tagged bike!
point, dest = random.sample(high_btw, 2)
tagged_bike = Bike(point=point, dest=dest, router=router)
tagged_bike.set_mode('tagged')
tagged_bike.trail = []
# add it to the total bike population
N.append(tagged_bike)
Run simulation¶
for t in range(simulation_steps):
# update population logs
tagged_pop[t] = len([1 for b in N if b.mode == 't'])
untagged_pop[t] = len([1 for b in N if b.mode == 'u'])
for b in N:
b.play_tag()
# make a log entry every 100 steps
if t % 100 == 0:
b.trail.append({'point': (float(b.point().to_string()[1]),
float(b.point().to_string()[0])),
'mode': b.mode})
Plot population dynamics¶
fig = plt.figure()
ax = fig.gca()
plt.plot(range(simulation_steps), tagged_pop, color='green', label='tagged')
plt.plot(range(simulation_steps), untagged_pop, color='deeppink', label='untagged')
ax.set_ylabel('population size')
ax.set_xlabel('simulation steps')
ax.legend()
plt.show()
Plot some frames¶
Tagged agents are green, untagged agents are pink.
# get edges of street network into geodataframe, for easier re-use
gdf = ox.plot.graph_to_gdfs(G, nodes=False)
fig, axes = plt.subplots(nrows=6, ncols=4,
sharex='all', sharey='all',
figsize=(9, 12), dpi=120)
axes = axes.flat
for ax, t in zip(axes, range(24)):
# plot streets
gdf.plot(ax=ax, linewidth=0.2, color="grey")
# mark bases
ax.add_artist(plt.Circle((base_A['x'], base_A['y']), 0.001, color='red', alpha=0.7))
ax.add_artist(plt.Circle((base_B['x'], base_B['y']), 0.001, color='blue', alpha=0.7))
# mark agent locations at current timestep
for b in N:
if b.trail[t]['mode'] == 'u':
color = 'deeppink'
elif b.trail[t]['mode'] == 't':
color = 'green'
ax.add_artist(plt.Circle(b.trail[t]['point'], 0.00035, color=color, alpha=0.6))
ax.set_title("t=%s" % (t*100), fontdict={'fontsize': 8})
ax.set_xticks([])
ax.set_yticks([])
plt.show()
