다익스트라 알고리즘 (Dijkstra Algorithm) 정리와 구현

최단 경로 문제 (Shortest Path Problem)의 해법 중 하나인 다익스트라 알고리즘 (Djikstra Algorithm)은 에츠허르 다익스트라가 고안한 알고리즘이다.

본 글에서는 다익스트라 알고리즘을 구현하기 위해, 알고리즘을 정리하고 실습 환경을 구성하여 직접 알고리즘을 구현해본다.

 

 

다익스트라 알고리즘의 나만의 요약

시작 지점에서 가본 곳 중에서 가장 가까운 곳(비용이 덜 드는 곳)부터 수색해나간다.

 

알고리즘

초기화 동작

1. 출발점과 도착점을 정의한다.

2. 출발점으로부터 모든 꼭지점(정점, Vertex)과의 최단 거리 값을 저장할 최단거리 테이블에 대해서 출발점을 0으로, 다른 모든 꼭짓점을 무한대 값으로 초기화한다.

3. 출발점에서 어떤 점으로 가장 빠르게(최단 거리로) 갈 수 있다고 계산한 직전 점 테이블을 초기화 한다.

4. 모든 점을 미방문 상태로 표시한다.

반복 동작

3-1. 현재점에서 갈 수 있는 이웃한 어떤 점에 대해서, 최단거리 테이블에 저장하고 있는 어떤 점의 거리값과 현재점을 거쳐서 어떤 점을 갈 때의 거리 값을 비교하여 최단거리인 값을 최단거리 테이블에 저장한다. (이때 방문해야할 이웃 점을 우선순위 큐에 올린다.)

3-2. 이웃한 어떤 점까지의 최단거리인 이전 점을 직전 점 테이블에 저장한다.

4. 현재점의 이웃한 모든 점에 대해 3번 작업을 수행한다.

5. 현재점을 방문 상태로 바꾼다. 방문한 점은 이후에 다시 방문하지 않는다.

6. 최단거리 테이블로부터 출발점에서 최단 거리 값을 가지면서 미방문 상태인 점을 찾는다. (우선순위 큐에서 최단거리인 미방문 이웃 점을 가져온다)

반복 종료 조건

7-1. 모든 점을 방문했다면 멈춘다.

7-2. 도착 점이 방문상태로 바뀌었다면 멈춘다.

 

Note.

• 최단거리 테이블은 출발 점에서 어떤 점까지의 최단거리 값을 저장한다.
• 직전 점 테이블은 출발 점에서 시작했을 때, 경로 상 현재 점 직전의 점을 가리킨다. 연결 리스트의 포인터와 같다.

 

구현 결과

격자 공간의 규칙은 대각선을 포함한 인접한 칸으로 모두 이동이 가능하다고 정의했으며, 거리 값은 직선 거리로 하여 구현이 용이하게 했다.

아래의 구현결과에서 표시는 다음과 같다.

회색 박스 : 도달 불가

초록 박스 : 시작 점

빨간 박스 : 도착 점

흰색 박스 : 미방문 점

노란 박스 : 방문한 점

파란 박스 : 탐색 큐에 있는 점

파란 점선 : 탐색 큐 점들의 최단 경로

 

동작 샘플 1

공간이 작으면 빨리 찾는다.

동작 샘플 2

도달할 수 없는 점에 대해서, 탐색 가능한 이웃 점이 없기 때문에 탐색 큐가 비어서 종료된다.

동작 샘플 3

하다보면 알고리즘이 느려지는 것 같지만 방문 점과 인접한 미방문 점을 찾고 경로를 그리는게 좀 걸린다.

그리고 특정 방향성이 있다기 보다는 출발점으로부터 최단거리인 점을 우선순위 큐에서 찾아서 쓰기 때문에, 중심에서 퍼져나가는 듯한 경향성을 보인다.

 

알고리즘의 구현

격자 세계의 거동을 기술하고 가시화하는 field_maker 클래스와 dijkstra 알고리즘을 실행하는 dijkstra 클래스를 만들었다. 또한 예제 맵을 만들기 위해 map_sample.py 를 따로 두었다.

아래 항목은 각각의 파일이며, 접은 글을 열어보면 코드를 볼 수 있다.

 

구현의 특징

• 모든 공간(field)은 방문, 미방문, 접근 불가 점 중에 하나의 상태를 가지기 때문에 공간 정보 값에 방문, 미방문, 접근 불가 정보를 같이 넣었다.
• 방문 점 근방의 미방문 점에 대해서 탐색 큐에 넣고, 우선순위 큐 (PriorityQueue)를 이용하여 6번 단계를 구현했다.
• 우선순위 큐에 탐색할 점을 넣어서 쓰다보니, 정작 탐색해야할 전체 점 정보를 얻어오기 어려워서 탐색 우선순위 큐(search_q)와 탐색 점 집합 (search_points)으로 나누어 변수를 쓰게 되었다.
• 우선순위 큐에서 pop 하여 현재점을 선택하면 탐색 점 집합에서 빼버렸는데, 탐색 점 집합으로 경로를 계산하다보니 도착 점에 도달하면 이걸 무시해버리는 문제가 있었다. 도착 점이라면 다시 넣도록 했다.
• drawnow 사용 시, 함수 내부에서 figure 선언을 하지 않으면 Figure가 업데이트 되지 않는 문제가 있다.

 

1. main_dijkstra.py

field_maker 와 dijkstra 클래스의 인스턴스가 상호작용하는 메인 코드

import matplotlib.pyplot as plt
import matplotlib.patches as patches
import numpy as np
from drawnow import *
import os
from map_sample import map_sample
 
from field_maker import FieldMaker
from algorithm_dijkstra import Dijkstra
 
if __name__ == "__main__":
    # Initialization of Field
    idx = 2
    p_start = map_sample[idx]['p_start']
    p_end   = map_sample[idx]['p_end']
    width   = map_sample[idx]['width']
    height  = map_sample[idx]['height']
    field   = map_sample[idx]['field']
    
    field_maker = FieldMaker(p_start, p_end, field)
    dijkstra    = Dijkstra(p_start, p_end, field)
 
    # Sample Path and Title
    path_list = list()
    title_str = "{}th".format(1)
    drawnow(field_maker.drawnow)
    # os.system('pause')
 
    # Search Shortest Path and Generate Path line
    for i in range(300):
        print("{}th Search Q size : {}".format(i+1, dijkstra.get_search_q_length()))
        now_point = dijkstra.get_priority_point()
        print("Priority Point ", now_point)
 
        # 3-1
        # Get neighborhood points
        neighbors = field_maker.get_neighborhood(now_point[0], now_point[1])
        field_maker.set_visited(now_point[0], now_point[1])
 
        # Search neighborhood points
        for neighbor in neighbors:
            dijkstra.search_nearby(neighbor)
 
        # Update path list
        path_list = dijkstra.get_pathlist()
 
        # Draw Map
        title_str = "{:4d}th @ ({:2d}.{:2d})".format(i, now_point[0], now_point[1])
        field_maker.set_title(title_str)
        field_maker.set_path_list(path_list)
        drawnow(field_maker.drawnow)
 
        # Termination Condition
        if(dijkstra.is_it_end() != 0):
            break
    
    # Final Path
    if(dijkstra.is_it_end() == 1): # Success
        path_list.clear()
        path = dijkstra.get_shortest_path()
        field_maker.set_shortest_path(path)
        path_list.append(path)
        field_maker.set_path_list(path_list)
 
    title_str = "End @ {}th - {}".format(i, dijkstra.get_end_reason())
    field_maker.set_title(title_str)
    drawnow(field_maker.drawnow)
    plt.show()

 

2. field_maker.py

field_maker 클래스를 정의한 파일

격자 공간의 거동을 정의하고, 격자 공간의 상태와 탐색 상태를 그려주는 코드

import matplotlib.pyplot as plt
import matplotlib.patches as patches
import numpy as np
from drawnow import *
 
class FieldMaker():
    def __init__(self, start_point, end_point, field):
 
        # self._field = np.flip(field, axis=1)
        self._field = field
        [width, height] = np.shape(self._field)
        self._height = height
        self._width = width
        # self._dist_table = np.zeros([height, width]) + width + height
        print("Field Shape : w {} h {}".format(width, height))
        # print(self._field)
 
        self._fig = plt.figure(1)
        self._ax = self._fig.add_subplot(1,1,1)
        self._ax.set_xticks(np.arange(-1.0, width+2,  1.0))
        self._ax.set_yticks(np.arange(-1.0, height+2, 1.0))
        self._start = start_point
        self._end   = end_point
 
        self._title = None
        self._path_list = None
        self._shortest_path = None
 
    def _draw_patch(self, x, y, facecolor, edgecolor=None, alpha=0.5):
        self._ax.add_patch(patches.Rectangle((x-0.5, y-0.5), 1, 1, facecolor=facecolor, edgecolor=edgecolor, fill=True, alpha=alpha))        
 
    def _draw_start_end_point(self):
        # Start-End Point
        self._draw_patch(self._start[0], self._start[1], 'green')
        self._draw_patch(self._end[0],   self._end[1],   'red')
 
    def _draw_field(self):
        # field
        for w in range(0, self._width):
            for h in range(0, self._height):
                if(w == self._start[0] and h == self._start[1]): # Start
                    continue
                if(w == self._end[0] and h == self._end[1]): # End
                    continue
                if(self._field[w,h] == 1):   # Blocked
                    self._draw_patch(w, h, facecolor='0.3', edgecolor='0.3')
                elif(self._field[w,h] == 2): # Visited
                    self._draw_patch(w, h, facecolor='yellow')
 
    def _draw_path_line(self, path_list):
        for path in path_list:
            self._ax.plot(path['x'], path['y'],':b', linewidth=2)
            self._draw_patch(path['x'][-1], path['y'][-1], facecolor='blue', alpha=0.1)
 
    def _draw_shortest_line(self, shortest_path):
        self._ax.plot(shortest_path['x'], shortest_path['y'],'r', linewidth=2)
        # print("Find Shortest Line : ", shortest_path)
 
    def _draw_grid(self):
        # Grid
        for w in range(0, self._width+1):
            self._ax.plot([w-0.5, w-0.5], [self._height-0.5, 0-0.5],'k', linewidth=2)
        for h in range(0, self._height+1):
            self._ax.plot([self._width-0.5, 0-0.5], [h-0.5, h-0.5], 'k', linewidth=2)
 
    def set_title(self, title):
        self._title = title
 
    def set_path_list(self, path_list):
        self._path_list = path_list
 
    def set_shortest_path(self, path):
        self._shortest_path = path
 
    def draw(self, path_list=None, title=None, shortest_path=None):
        self._fig = plt.figure(1)
        self._ax = self._fig.add_subplot(1,1,1)
        self._draw_grid()
 
        self._draw_start_end_point()
        self._draw_field()
 
        if(title != None):
            self._ax.set_title(title)
        if(path_list != None):
            self._draw_path_line(path_list)
        if(shortest_path != None):
            self._draw_shortest_line(shortest_path)
 
        self._ax.grid()
        self._ax.set_xlim([-1, self._width])
        self._ax.set_ylim([-1, self._height])
        self._ax.set_xlabel("X")
        self._ax.set_ylabel("Y")
 
    def drawnow(self):
        self.draw(self._path_list, self._title, self._shortest_path)
        
    def get_neighborhood(self, x, y):
        neighbor = list()
        for dx in range(-1, 2):
            for dy in range(-1, 2):
                px = x + dx
                py = y + dy
 
                if(px < 0 or px >= self._width): # Out of bound
                    continue
                if(py < 0 or py >= self._height): # Out of bound
                    continue
                if(dx == 0 and dy == 0): # Not Me
                    continue
                if(self._field[px][py] == 1): # field
                    continue
                if(self._field[px][py] == 2): # Visited
                    continue
                point = dict()
                point['x'] = px
                point['y'] = py
                point['dist'] = np.sqrt(dx*dx + dy*dy)
                neighbor.append(point)
 
        return neighbor
 
    def set_visited(self, x, y):
        self._field[x][y] = 2 # Visited
        # print("Set visited : ({}, {})".format(x,y))
 
    def get_field(self):
        field = np.flip(self._field, axis=1)
        return field.transpose()
 

 

3. algorithm_dijkstra.py

dijkstra 클래스를 정의한 파일

다익스트라 알고리즘을 직접적으로 푸는 파일

import numpy as np
from queue import PriorityQueue
 
class Dijkstra():
    def __init__(self, start_point, end_point, field):
 
        # Get Field Information
        #   Field should include blocked information
        self._field = field
        [width, height] = np.shape(self._field)
        self._height = height
        self._width = width
 
        # Define Start/End Point
        self._start = start_point
        self._end   = end_point
 
        # Shortest distance table
        self._dist_table  = np.zeros([self._width, self._height]) + self._width * self._height
 
        # Path-Tracing table
        self._prev = dict()
        self._prev['x']   = np.zeros([self._width, self._height]) - 1
        self._prev['y']   = np.zeros([self._width, self._height]) - 1
    
        self._search_q = PriorityQueue()
        self._search_points = dict()
 
        # Initial Starting Point
        self._search_q.put((0, self._start))
        self._dist_table[self._start[0]][self._start[1]] = 0
        self._prev['x'][self._start[0]][self._start[1]] = self._start[0]
        self._prev['y'][self._start[0]][self._start[1]] = self._start[1]
 
        self._visiting_point = None
        self._visiting_dist  = None
 
        self._did_find_shortest_path = False
        self._end_reason = ""
 
    def get_search_q_length(self):
        return self._search_q.qsize()
 
    def get_priority_point(self):
        '''
        return shortest x,y point in search queue
        '''
 
        priori_point = self._search_q.get()
        self._visiting_dist  = priori_point[0]
        self._visiting_point = priori_point[1]
 
        # Remove visiting point in search queue
        if ((self._visiting_point[0], self._visiting_point[1]) in self._search_points):
            del self._search_points[(self._visiting_point[0], self._visiting_point[1])]
 
        # If visiting point is end point, leave it in search queue
        if (self._visiting_point[0] == self._end[0] and self._visiting_point[1] == self._end[1]):
            self._search_points[(self._end[0], self._end[1])] = dict()
 
        self._field[self._visiting_point[0]][self._visiting_point[1]] = 2 # Visited
 
        return self._visiting_point # [x,y]
 
    def search_nearby(self, neighbor):
        '''
        Search nearby
        '''
        neighbor_a_step_dist = neighbor['dist']
        neighbor_dist = neighbor_a_step_dist + self._visiting_dist
        if(neighbor_dist < self._dist_table[neighbor['x']][neighbor['y']]):
            # Update shortest path and trace path
            self._dist_table[neighbor['x']][neighbor['y']] = neighbor_dist
            self._prev['x'][neighbor['x']][neighbor['y']] = self._visiting_point[0]
            self._prev['y'][neighbor['x']][neighbor['y']] = self._visiting_point[1]
 
            # Update search queue
            search_point = (neighbor_dist, [neighbor['x'], neighbor['y']])
            self._search_q.put(search_point)
            if ((neighbor['x'], neighbor['y']) not in self._search_points):
                self._search_points[(neighbor['x'], neighbor['y'])] = dict()
            # print(" new! ->", search_point)
 
    def get_pathlist(self):
        path_list = list()
        
        for key, value in self._search_points.items():
            path = self._get_path(self._start[0], self._start[1], key[0], key[1], self._prev)
            path_list.append(path)
 
        return path_list
    
    def get_shortest_path(self):
        if(self._did_find_shortest_path == True):
            path = self._get_path(self._start[0], self._start[1], self._end[0], self._end[1], self._prev)
            return path
        else:
            return None
 
    # Termination Condition
    def is_it_end(self):
 
        # Empty queue?
        if(self._search_q.empty() == True):
            print("[INFO] Empty Search Queue")
            self._end_reason = "Empty Search Queue"
            return 2 # Failed to find available points
        
        # End point is visitied ?
        if(self._field[self._end[0]][self._end[1]] == 2):
            self._did_find_shortest_path = True
            self._search_points[(self._end[0], self._end[1])] = dict()
            print("[INFO] Arrive End point ({}, {})".format(self._end[0], self._end[1]))
            self._end_reason = "Success"
            return 1 # Success 
        
        self._end_reason = "Failed to arrive"
        return 0
    
    def get_end_reason(self):
        return self._end_reason
    
    def _get_path(self, start_x, start_y, end_x, end_y, linker_table:dict):
        path = dict()
        path['x'] = list(); now_x = end_x
        path['y'] = list(); now_y = end_y
        path['x'].append(now_x)
        path['y'].append(now_y)
        while(True):
            # print("Now ", [now_x, now_y])
            # print(prev['x'])
            # print(prev['y'])
            if(now_x == start_x and now_y == start_y):
                break
            if(now_x == -1 or now_y == -1):
                break
            prev_x = int(linker_table['x'][now_x][now_y])
            prev_y = int(linker_table['y'][now_x][now_y])
            now_x = prev_x
            now_y = prev_y
            path['x'].append(prev_x)
            path['y'].append(prev_y)
        path['x'] = np.flip(path['x'], axis=0)
        path['y'] = np.flip(path['y'], axis=0)
 
        return path

 

4. map_sample.py

예제 맵을 사용하는 정의한 파일이다.

import numpy as np
 
map_sample = list()
 
# 0 : NULL
# 1 : field
# 2 : Visited
 
# Case 1
sample = dict()
sample['p_start'] = [ 0, 0]
sample['p_end']   = [ 6, 0]
sample['width']   = 8 # x 
sample['height']  = 5 # y
sample['field']   = np.array([  [0, 0, 0, 0, 0, 0, 0, 0],
                                [0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 1, 0, 0, 0, 0, 0]])
sample['field'] = sample['field'].transpose()
sample['field'] = np.flip(sample['field'], axis=1)
map_sample.append(sample)
# print(map_sample)
 
 
# Case 2 - Never arrival
sample = dict()
sample['p_start'] = [ 0, 0]
sample['p_end']   = [ 7, 0]
sample['width']   = 8 # x 
sample['height']  = 5 # y
sample['field']   = np.array([  [0, 0, 0, 0, 0, 0, 0, 0],
                                [0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 1, 0, 0, 0, 1, 1],
                                [0, 0, 1, 0, 0, 0, 1, 0]])
sample['field'] = sample['field'].transpose()
sample['field'] = np.flip(sample['field'], axis=1)
map_sample.append(sample)
# print(map_sample)
 
 
# Case 3
sample = dict()
sample['p_start'] = [ 0, 0]
sample['p_end']   = [19, 5]
sample['width']   = 20
sample['height']  = 12
sample['field']   = np.array(  [[0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                                [0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                                [0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
                                [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0]])
sample['field'] = sample['field'].transpose()
sample['field'] = np.flip(sample['field'], axis=1)
map_sample.append(sample)
# print(map_sample)
 
if __name__ == "__main__":
    print(map_sample[0])
    print(map_sample[1])

 

 

[1] E.W.  Dijkstra,  “A  note  on  two  problems  in  connection  with  graphs,” Numerische  Mathmatik,  1959,  1:269-271, Available at https://ir.cwi.nl/pub/9256/9256D.pdf

[1] Lee Kyunghan, "Dijkstra's Algorithms," Lecture Note, Available at https://ocw.snu.ac.kr/sites/default/files/NOTE/W16.Dijkstra.pdf

 

 

 

 

*** EOF ***