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


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'


  1. 50 buildings in the area surrouding the University of Toronto were selected as destinations, or “points of emergency”.

  2. 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).


  1. Only one request at a time.

  2. 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.

  3. The three emergency levels are high, medium, and low.

  4. 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 = 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):

    def load_vehicles(self, filename="emergency_vehicle_location.csv"):
        df = pd.read_csv(filename)
        for index, row in df.iterrows():
                    (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():

    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)
                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)

            # 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(
            key=lambda x: haversine_distance(
        while level > 0:
            current = available_vehicles.pop(0)
            level -= current.capacity

        results = copy(dispatched)
        if level < 0:
            # Try to remove some vehicles
            for v in dispatched:
                if level + v.capacity <= 0:
                    level += v.capacity
                if level == 0:

        return emergency, results

    def visualize(self, G, result):

        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"
                    icon=icons[v.vehicle_type - 1], 
                popup=f"Capacity: {v.capacity}"

        # Destination Marker
            icon=folium.Icon(color="orange", icon="exclamation", prefix="fa"),
            popup=f"Level: {result['emergency'].level}"
        # Routes
        for r in result['routes']:
        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)


Vehicles are dispatched as follows:

  1. 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.

  2. The Haversine distance between the remaining vehicles and the emergency location is determined, and the list is sorted in ascending order.

  3. 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.


# 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)

                children = []
                children_pheremones = []
                for child in parent.expand():
                    # If we see the destination, ignore all pheremones
                    if child == destination:
                        found = True
                        route = child.path()
                    if child not in explored:
                                pheremone_concentrations[(parent.osmid, child.osmid)],

                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

                # 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
                known_routes[route] = 1
        return list(max(known_routes, key=known_routes.get))

Results and Visualization#

[Level:2, Type:3, Building: building_2]
Number of Vehicles:1
Total Distance: 327.496
Total Turns: 2
Make this Notebook Trusted to load map: File -> Trust Notebook
[Level:2, Type:1, Building: building_13]
Number of Vehicles:1
Total Distance: 2131.913
Total Turns: 19
Make this Notebook Trusted to load map: File -> Trust Notebook
[Level:1, Type:1, Building: building_6]
Number of Vehicles:1
Total Distance: 773.132
Total Turns: 5
Make this Notebook Trusted to load map: File -> Trust Notebook

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 = 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:
                    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(
                    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#

[Level:2, Type:1, Building: building_34]
Number of Vehicles:1
Total Distance: 1133.353
Total Turns: 7
Make this Notebook Trusted to load map: File -> Trust Notebook
[Level:3, Type:2, Building: building_7]
Number of Vehicles:1
Total Distance: 1401.385
Total Turns: 8
Make this Notebook Trusted to load map: File -> Trust Notebook
[Level:2, Type:2, Building: building_22]
Number of Vehicles:1
Total Distance: 544.236
Total Turns: 3
Make this Notebook Trusted to load map: File -> Trust Notebook

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.