Emergency Dispatch and Routing#
Authors: Tianjian Chen, Aswin Raj Giri and Vishnu Priya Rajendran
Course: ECE1724H: Bio-inspired Algorithms for Smart Mobility - Fall 2021
Instructor: Dr. Alaa Khamis
Department: Edward S. Rogers Sr. Department of Electrical & Computer Engineering, University of Toronto
Introduction#
As cities and grow and the number of cars on roads increase, accidents and emergency situations will also continue to increase. This makes the problem of dispatching and routing emergency vehicles even more critical.
Dispatching is the concept of allotting and sending the nearby available vehicles to the location of an emergency, while routing deals with selecting the ideal route to reach that destination. This particular example places an emphasis on incident response time, which is considered as the primary cost function. The incident response time is approximated using the route distance and the number of turns (as turning generally takes more time).
Two example solutions are provided, using Ant Colony Optimization and the Firefly Algorithm.
import osmnx
from smart_mobility_utilities.common import Node, cost, randomized_search
from smart_mobility_utilities.problem import haversine_distance, cross_over
import math
import random
from tqdm.notebook import tqdm
import osmnx as ox
from numpy.random import choice as weighted_choice
import pandas as pd
from copy import copy, deepcopy
import folium
---------------------------------------------------------------------------
ModuleNotFoundError Traceback (most recent call last)
Input In [1], in <cell line: 2>()
1 import osmnx
----> 2 from smart_mobility_utilities.common import Node, cost, randomized_search
3 from smart_mobility_utilities.problem import haversine_distance, cross_over
4 import math
ModuleNotFoundError: No module named 'smart_mobility_utilities'
Datasets#
50 buildings in the area surrouding the University of Toronto were selected as destinations, or “points of emergency”.
The real locations of 8 fire stations, 5 police stations, and 10 hospitals in that area were selected as vehicle depots for various types of emergency vehicles (fire trucks, police cars, ambulances).
Constraints#
Only one request at a time.
Emergency sites have an associated “emergency level” and emergency vehicles have a “capacity” to satisfy those levels. Emergency sites can only be satisfied by a vehicle of an equal or higher capacity, or by multiple vehicles whose sum meets or exceeds the emergency site’s emergency level.
The three emergency levels are high, medium, and low.
All emergency vehicles are initialized at their starting depot.
Functions and Classes#
Cost Function#
Turns in Route#
This function calculates the number of turns in a route between two nodes as the number of unique street ids - 1.
# Determines the number of turns in a route between two locations
def turns(G,route):
roads = []
for i in range(0,len(route)-1):
way =G[route[i]][route[i+1]]
roads.append(str(way[0]['osmid']))
roads = len(set(roads))
turn = roads -1
return turn
Closest Node#
This function looks for the OSMID of the closest node in an OSMNX graph for a given geocode.
def closest_node_osmid(G, lat, lng):
id = ox.distance.nearest_nodes(G,lng,lat)
return Node(G,id)
Emergency Class#
class Emergency:
def __init__(self, building, level, type):
self.level = level
self.type = type
self.building = building
def __repr__(self):
return f"[Level:{self.level}, Type:{self.type}, Building: {self.building.name}]"
Vehicle Class#
class Vehicle:
def __init__(self, name, capacity, vehicle_type, location):
self.name = name
self.capacity = capacity
self.vehicle_type = vehicle_type
self.location = location
def __repr__(self):
return f"[Name:{self.name}, Type:{self.vehicle_type}, Capacity: {self.capacity}, Location: {self.location}]"
Building Class#
class Building:
def __init__(self, name, lat, lng, level):
self.name = name
self.location = (lat, lng)
self.level = level
Problem Class#
class EmergencyRouting:
vehicles = []
locations = []
emergencies = []
results = []
def __init__(self):
self.load_vehicles()
self.load_buildings()
def load_vehicles(self, filename="emergency_vehicle_location.csv"):
df = pd.read_csv(filename)
for index, row in df.iterrows():
self.vehicles.append(
Vehicle(
row.name,
row.casualities_expected_to_treat,
row.vehicle_type,
(row.vehicle_lat, row.vehicle_lng),
)
)
def load_buildings(self, filename="building_location.csv"):
df = pd.read_csv(filename)
for index, row in df.iterrows():
self.locations.append(
Building(
row.building_name,
row.building_lat,
row.building_lng,
row.casualities_expected,
)
)
def generate_emergencies(self, num_emergencies, allow_repeat_building=False):
results = []
buildings = copy(self.locations)
for i in range(num_emergencies):
# Select a random emergency level, weighted towards less severe ones
e_level = weighted_choice([1, 2, 3], 1, [0.5, 0.3, 0.2])[0]
# Select a random emergency type
e_type = random.choice([1, 2, 3])
# Select a random building
if allow_repeat_building:
e_building = random.choice(buildings)
else:
e_building = buildings.pop(random.randrange(len(buildings)))
results.append(Emergency(e_building, e_level, e_type))
self.emergencies = results
def route(self, G, vehicles, destination, algorithm):
routes = []
destination_node = closest_node_osmid(G, destination[0], destination[1])
for v in vehicles:
origin_node = closest_node_osmid(G, v.location[0], v.location[1])
route = algorithm.run(G, origin_node, destination_node)
routes.append(route)
# Update vehicle location
v.location = (destination[0], destination[1])
self.vehicles[self.vehicles.index(v)] = v
return routes
# Takes the next emergency and assigns vehicles to be dispatched
def dispatch(self):
dispatched = []
emergency = self.emergencies.pop(0)
level = emergency.level
# Filter to only keep relevant vehicles
available_vehicles = [
v for v in self.vehicles if v.vehicle_type == emergency.type
]
available_vehicles = sorted(
available_vehicles,
key=lambda x: haversine_distance(
emergency.building.location[1],
emergency.building.location[0],
x.location[1],
x.location[0],
),
)
while level > 0:
current = available_vehicles.pop(0)
dispatched.append(current)
level -= current.capacity
results = copy(dispatched)
dispatched.reverse()
if level < 0:
# Try to remove some vehicles
for v in dispatched:
if level + v.capacity <= 0:
results.pop(results.index(v))
level += v.capacity
if level == 0:
break
return emergency, results
def visualize(self, G, result):
print(result["emergency"])
print(f"Number of Vehicles:{len(result['active_vehicles'])}")
print(f"Total Distance: {sum([cost(G,r) for r in result['routes']])}")
print(f"Total Turns: {sum([turns(G,r) for r in result['routes']])}")
icons = ["fire-extinguisher", "gavel", "ambulance"]
v_colors = ["red", "blue", "green"]
m = folium.Map((43.654589, -79.389229), zoom_start=10)
# Vehicle Markers
for v in result["vehicle"]:
color = (
v_colors[v.vehicle_type - 1]
if v.name in result["active_vehicles"]
else "gray"
)
folium.Marker(
location=v.location,
icon=folium.Icon(
color=color,
icon=icons[v.vehicle_type - 1],
prefix="fa"
),
popup=f"Capacity: {v.capacity}"
).add_to(m)
# Destination Marker
folium.Marker(
location=result["emergency"].building.location,
icon=folium.Icon(color="orange", icon="exclamation", prefix="fa"),
popup=f"Level: {result['emergency'].level}"
).add_to(m)
# Routes
for r in result['routes']:
ox.plot_route_folium(G,r,m)
return m
def run(self, G, num_emergencies, allow_repeat_locations):
# Generate emergencies
self.generate_emergencies(num_emergencies, allow_repeat_locations)
def next(self, algorithm):
if len(self.emergencies) == 0:
return "No more emergencies to respond to."
# Dispatch
emergency, vehicles = self.dispatch()
result_obj = {
"emergency": emergency,
"vehicle": deepcopy(self.vehicles),
"active_vehicles": [x.name for x in vehicles]
}
# Route
routes = self.route(G, vehicles, emergency.building.location, algorithm)
result_obj["routes"] = routes
return self.visualize(G, result_obj)
Dispatching#
Vehicles are dispatched as follows:
The list of all vehicles is filtered to remove any vehicles that do not match the emergency type. For example, ambulances will not be considered when responding to a fire call.
The Haversine distance between the remaining vehicles and the emergency location is determined, and the list is sorted in ascending order.
Vehicles are added to the dispatch list until the severity level of the emergency is satisfied. If the final level of the dispatched fleet is greater than what is required, we will try to remove low level vehicles to match the emergency level perfectly.
For example:
If there is an emergency of level 3, and the closest vehicles are:
A (level 1): 2 km away
B (level 1): 4 km away
C (level 3): 6 km away
The dispatching algorithm will select all 3 vehicles (A, B, C), but then remove both A and B as they are not required to respond to the emergency.
Solutions#
# Initialize problem
problem = EmergencyRouting()
G = osmnx.graph_from_address("University Of Toronto", dist=3000, network_type="drive_service")
# Initialize some parameters
num_emergencies = 9
allow_repeat_locations = False
problem.run(G, num_emergencies, allow_repeat_locations)
Example 1: Dispatching + Routing Emergencies using ACO#
Ant Colony Optimization#
class ACO:
def __init__(self, alpha=2, beta=2, n=500, Q=1):
self.alpha = alpha
self.beta = beta
self.n = n
self.Q = Q
def pheremone(self, level, distance):
return level ** self.alpha * ((1/distance)) ** self.beta
def run(self, G, origin, destination):
# randomize the pheromones
pheremone_concentrations = {(u,v):random.uniform(0,0.5) for [u,v] in G.edges()}
known_routes = dict()
for ant in tqdm(range(self.n)):
# Place the ant at the colony
frontier = [origin]
explored = set()
route = []
found = False
while frontier and not found:
parent = frontier.pop(0)
explored.add(parent)
children = []
children_pheremones = []
for child in parent.expand():
# If we see the destination, ignore all pheremones
if child == destination:
found = True
route = child.path()
continue
if child not in explored:
children.append(child)
children_pheremones.append(
self.pheremone(
pheremone_concentrations[(parent.osmid, child.osmid)],
child.distance
)
)
if len(children) == 0:
continue # The ant is stuck, go back.
transition_probability = [
children_pheremones[i] / sum(children_pheremones)
for i in range(len(children_pheremones))
]
# Probabilistically choose a child to explore based weighted by transition probability
chosen = random.choices(children, weights=transition_probability, k=1)[0]
# Add all the non-explored children in case we need to explore them later
children.pop(children.index(chosen))
frontier.extend(children)
# Set the chosen child to be the next node to explore
frontier.insert(0, chosen)
# We now have a completed route, we can increase pheremone levels
# on that route for the next ant to detect.
for u, v in zip(route[:-1], route[1:]):
length_of_edge = G[u][v][0]['length']
pheremone_concentrations[(u,v)] += self.Q/length_of_edge
# If the route is newly discovered, add it to the list
route = tuple(route)
if route in known_routes:
known_routes[route] += 1
else:
known_routes[route] = 1
return list(max(known_routes, key=known_routes.get))
Results and Visualization#
problem.next(ACO())
[Level:2, Type:3, Building: building_2]
Number of Vehicles:1
Total Distance: 327.496
Total Turns: 2
problem.next(ACO())
[Level:2, Type:1, Building: building_13]
Number of Vehicles:1
Total Distance: 2131.913
Total Turns: 19
problem.next(ACO())
[Level:1, Type:1, Building: building_6]
Number of Vehicles:1
Total Distance: 773.132
Total Turns: 5
As you can see from the above three examples, the ACO solver does not take into account the number of turns in a route, as pheremones are deposited on node-to-node edges. Since the scoring of routes in ACO is the sum of individual scores, we cannot penalize excess turns.
Example 2: Dispatching + Routing Emergencies using Firefly Algorithm#
Firefly Algorithm Class#
class Firefly:
def __init__(self, num_iterations=15, pop_size=100, gamma=2, k=100):
self.num_iterations = num_iterations
self.pop_size = pop_size
self.gamma = gamma
self.k = k
# Cost as length + complexity of route
def turns(G,route):
roads = []
for i in range(0,len(route)-1):
way =G[route[i]][route[i+1]]
roads.append(str(way[0]['osmid']))
roads = len(set(roads))
turn = roads -1
return turn
def luminosity(self, G, route):
return cost(G,route) + self.k*turns(G,route)
# Define the distance between two routes as the number of common nodes they possess
def distance(self, route1,route2):
return len(set(route1) & set(route2))
def run(self, G, origin, destination):
# Initialize population
flies = [randomized_search(G,origin.osmid, destination.osmid) for _ in range(self.pop_size)]
for _ in tqdm(range(self.num_iterations)):
for i in range(self.pop_size):
flies_luminosity = list() # for all flies except i
for j in range(self.pop_size):
if i == j:
continue
flies_luminosity.append((j, self.luminosity(G, flies[j])))
moving_fly = flies[i]
# Using min here as the "most" luminous is the one with lowest cost + complexity
target_fly = min(
flies_luminosity,
key=lambda fly: fly[1]
* math.exp(-1 * self.gamma * self.distance(moving_fly, flies[fly[0]]))
)
target_fly = flies[target_fly[0]]
# update the position, moving is just a crossover
flies[i] = cross_over(target_fly,moving_fly)
return min(flies, key=lambda fly: self.luminosity(G,fly))
Results and Visualization#
problem.next(Firefly())
[Level:2, Type:1, Building: building_34]
Number of Vehicles:1
Total Distance: 1133.353
Total Turns: 7
problem.next(Firefly())
[Level:3, Type:2, Building: building_7]
Number of Vehicles:1
Total Distance: 1401.385
Total Turns: 8
problem.next(Firefly())
[Level:2, Type:2, Building: building_22]
Number of Vehicles:1
Total Distance: 544.236
Total Turns: 3
As you can see, the results generated by the Firefly algorithm produce routes with fewer turns, as a penalty was applied to the cost function based on the number of turns in a route.