From 0e90839ea0aa59695f8c23d305eb3e6a85a519c1 Mon Sep 17 00:00:00 2001
From: "niehues.mark@gmail.com" <niehues.mark@gmail.com>
Date: Fri, 13 Mar 2020 12:51:00 +0100
Subject: [PATCH] adding priority queue implementation
---
evrouting/utils.py | 44 ++++++++++++++++++++++++++++++++++++++++++++
tests/test_utils.py | 33 +++++++++++++++++++++++++++++++++
2 files changed, 77 insertions(+)
create mode 100644 evrouting/utils.py
create mode 100644 tests/test_utils.py
diff --git a/evrouting/utils.py b/evrouting/utils.py
new file mode 100644
index 0000000..e191009
--- /dev/null
+++ b/evrouting/utils.py
@@ -0,0 +1,44 @@
+import itertools
+from heapq import *
+
+
+class PriorityQueue:
+ REMOVED = '<removed-task>' # placeholder for a removed task
+
+ def __init__(self):
+ self.pq = [] # list of entries arranged in a heap
+ self.entry_finder = {} # mapping of tasks to entries
+ self.counter = itertools.count() # unique sequence count as tie break
+
+ def insert(self, item, priority=0):
+ """Add a new task or update the priority of an existing task"""
+ if item in self.entry_finder:
+ self.remove_item(item)
+ count = next(self.counter)
+ entry = [priority, count, item]
+ self.entry_finder[item] = entry
+ heappush(self.pq, entry)
+
+ def remove_item(self, item):
+ """Mark an existing task as REMOVED. Raise KeyError if not found."""
+ entry = self.entry_finder.pop(item)
+ entry[-1] = self.REMOVED
+
+ def delete_min(self):
+ """Remove and return the lowest priority task. Raise KeyError if empty."""
+ while self.pq:
+ priority, count, item = heappop(self.pq)
+ if item is not self.REMOVED:
+ del self.entry_finder[item]
+ return item
+ raise KeyError('pop from an empty priority queue')
+
+ def peak_min(self):
+ """Return minimum item without removing it from the queue."""
+ while self.pq:
+ priority, count, item = self.pq[0]
+ if item is not self.REMOVED:
+ return item
+ else:
+ heappop(self.pq)
+ raise KeyError('Empty queue.')
diff --git a/tests/test_utils.py b/tests/test_utils.py
new file mode 100644
index 0000000..2015392
--- /dev/null
+++ b/tests/test_utils.py
@@ -0,0 +1,33 @@
+import pytest
+from evrouting.utils import PriorityQueue
+
+
+@pytest.fixture
+def q():
+ q = PriorityQueue()
+ q.insert('high', 3)
+ q.insert('low', 2)
+
+ yield q
+ del q
+
+
+class TestProrityQueue:
+ def test_insert(self, q):
+ assert q.delete_min() == 'low'
+ assert q.delete_min() == 'high'
+
+ with pytest.raises(KeyError):
+ q.delete_min()
+
+ def test_peak(self, q):
+ assert q.peak_min() == 'low'
+ assert q.peak_min() == 'low' # does not get removed
+
+ def test_update(self, q):
+ q.insert('high', 1)
+ assert q.delete_min() == 'high'
+ assert q.delete_min() == 'low'
+
+ with pytest.raises(KeyError):
+ q.delete_min()
--
GitLab