Fully Dynamic Algorithms for Maintaining Shortest Paths Trees 리뷰 #3 - Decreasing the Weight of an Edge 구현

안녕하세요. 오늘은 오랜만에 와이프랑 계룡산에 다녀왔습니다.

햇살도 아름답고 꽃도 이쁘게 피어 있어서 사진을 왕창 찍었습니다 ㅎㅎ

 

여러분은 아름다운 봄을 즐기고 있으신가요?

 

이번에는 Fully Dynamic Algorithms for Maintaining Shortest Paths Trees의 Decreasing the Weight of an Edge 부분을 구현해봤습니다.

단편적으로 코드를 보여드리면 다음과 같습니다.

1. Decreasing the Weight of an Edge 구현

from typing import Dict, Any
import heapq
from dataclasses import dataclass, field

from common import (
    Graph,
    Edge,
)
from single_agent.output_information import (
    OutputInformation,
)


def decrease_edge_weight(
    graph: Graph,
    output_info_dict: Dict[int, OutputInformation],
    edge_id: int,
    amount: int,
):
    edge = graph.get_edge(edge_id=edge_id)
    if amount >= edge.weight:
        raise Exception("edge.weight cannot be zero or negative value")

    edge.set_weight(edge.weight - amount)

    for source_node in graph.nodes.values():
        decrease_edge_weight_for_single_source(
            graph,
            output_info_dict[source_node.id],
            edge,
        )


@dataclass(order=True)
class PrioritizedItem:
    priority: float
    item: Any = field(compare=False)


def decrease_edge_weight_for_single_source(
    graph: Graph,
    output_info: OutputInformation,
    edge: Edge,
):
    start_node = edge.start_node
    end_node = edge.end_node

    current_distance = output_info.get_distance(end_node.id)
    new_distance = output_info.get_distance(start_node.id) + edge.weight
    if new_distance >= current_distance:
        return

    output_info.set_distance(end_node.id, new_distance)
    output_info.update_parent_of_node(node_id=end_node.id, parent_id=start_node.id)

    queue = [
        PrioritizedItem(
            priority=output_info.get_distance(end_node.id),
            item=end_node,
        ),
    ]
    queue_dict = {end_node.id: queue[0]}
    while len(queue) > 0:
        current_item = heapq.heappop(queue)
        current_node = current_item.item
        del queue_dict[current_node.id]

        for edge in graph.get_adjacent_edges(current_node.id):
            current_distance = output_info.get_distance(edge.end_node.id)
            new_distance = output_info.get_distance(current_node.id) + edge.weight
            if new_distance >= current_distance:
                continue

            output_info.set_distance(edge.end_node.id, new_distance)
            output_info.update_parent_of_node(
                node_id=edge.end_node.id, parent_id=current_node.id
            )

            if edge.end_node.id not in queue_dict:
                item = PrioritizedItem(
                    priority=output_info.get_distance(edge.end_node.id),
                    item=edge.end_node,
                )
                heapq.heappush(queue, item)
                queue_dict[edge.end_node.id] = item
            else:
                queue_dict[edge.end_node.id].item.priority = output_info.get_distance(
                    edge.end_node.id
                )

• 참고: https://github.com/goodahn/path-finding/blob/main/single_agent/dynamic_problem.py
• 전체 코드는 다음 github repository에서 확인하실 수 있습니다. https://github.com/goodahn/path-finding

 

2. 결과

floyd warshall 알고리즘과 실행 시간을 비교해보면 엄청난 차이가 나긴 하더라고요 ㄷㄷㄷ

실행 결과 예시

 

아직 edge weight의 decrease만 구현했는데 increase도 구현하면

실제로 활용하고 싶은 시나리오에서 시간 차이가 얼마나 벌어지는지 확인해보려고 합니다!

 

🎁 Fully Dynamic Algorithms for Maintaining Shortest Paths Trees 이전 리뷰들 🎁

• Fully Dynamic Algorithms for Maintaining Shortest Paths Trees 리뷰 #2 - 중간 점검 output information 구현해보기
• Fully Dynamic Algorithms for Maintaining Shortest Paths Trees 리뷰 #1 - 논문을 읽게 된 이유 및 용어 정리