forked from hofmannol/AlgoDatSoSe25
Implemented a hacky PriorityQueue using MemoryArrays
This commit is contained in:
parent
62d6fa7459
commit
0401aa42ee
@ -1,12 +1,12 @@
|
||||
from utils.memory_array import MemoryArray
|
||||
from utils.memory_cell import MemoryCell
|
||||
from utils.literal import Literal
|
||||
from utils.constants import MIN_VALUE
|
||||
from utils.memory_manager import MemoryManager
|
||||
from utils.constants import MAX_VALUE
|
||||
from utils.memory_range import mrange
|
||||
|
||||
# Impl of MemoryArray says we cant add our own Datatypes beside Literal and List
|
||||
# BUUUUT we can just wrap our Datatype in a List :-)
|
||||
# We store them in a MemoryArray internaly tho anyhow so we increment our Counters for the RAM
|
||||
class HeapEntry:
|
||||
def __init__(self, item, priority=1):
|
||||
self.data = MemoryArray(Literal(2))
|
||||
@ -46,6 +46,10 @@ class HeapEntry:
|
||||
class PriorityQueue:
|
||||
def __init__(self, max_size : Literal = Literal(100)):
|
||||
self.heap = MemoryArray(max_size)
|
||||
# Add uninitialized HeapEntry Values so the Adds/Compares do not fail on emtpy stack.
|
||||
# Would have to switch to MIN_VALUE if we switch what is a "Higher" Prio
|
||||
for i in mrange(max_size.value):
|
||||
self.heap[i].set([HeapEntry(MAX_VALUE, MAX_VALUE)])
|
||||
self.size = MemoryCell(0)
|
||||
|
||||
def parent(self, i: Literal) -> Literal:
|
||||
@ -57,26 +61,27 @@ class PriorityQueue:
|
||||
def rightChild(self, i: Literal) -> Literal:
|
||||
return MemoryCell(MemoryCell(2) * i) + Literal(2)
|
||||
|
||||
# Swap the Lists -> Therefore get the value which is the List and then Set it again
|
||||
def swap(self, i: Literal, j: Literal):
|
||||
tmp_i = self.heap[i].value
|
||||
tmp_j = self.heap[j].value
|
||||
self.heap[i].set(tmp_j)
|
||||
self.heap[j].set(tmp_i)
|
||||
|
||||
def max_heapify(self, i: Literal):
|
||||
def maxHeapify(self, i: Literal):
|
||||
left = self.leftChild(i)
|
||||
right = self.rightChild(i)
|
||||
largest = i
|
||||
|
||||
if left < self.size and self.heap[left] > self.heap[largest]:
|
||||
if left < Literal(self.size.value) and self.heap[left].value[0] > self.heap[largest].value[0]:
|
||||
largest = left
|
||||
|
||||
if right < self.size and self.heap[right] > self.heap[largest]:
|
||||
if right < Literal(self.size.value) and self.heap[right].value[0] > self.heap[largest].value[0]:
|
||||
largest = right
|
||||
|
||||
if largest != i:
|
||||
self.swap(i, largest)
|
||||
self.max_heapify(largest)
|
||||
self.maxHeapify(largest)
|
||||
|
||||
def insert(self, entry : HeapEntry):
|
||||
if self.size >= self.heap.length():
|
||||
@ -86,139 +91,60 @@ class PriorityQueue:
|
||||
self.heap[i].set([entry])
|
||||
self.size += Literal(1)
|
||||
|
||||
currentNode = self.heap[i]
|
||||
parentNode : Literal = self.heap[self.parent(i)]
|
||||
print(currentNode)
|
||||
print(parentNode)
|
||||
|
||||
while i > Literal(0) and self.heap[self.parent(i)] < self.heap[i]:
|
||||
while i > Literal(0) and self.heap[self.parent(i)].value[0] < self.heap[i].value[0]:
|
||||
self.swap(i, self.parent(i))
|
||||
i = self.parent(i)
|
||||
|
||||
def pop(self):
|
||||
if self.is_empty():
|
||||
if self.isEmpty():
|
||||
raise IndexError("Queue is empty!")
|
||||
|
||||
max_item = self.heap[Literal(0)]
|
||||
max_item = self.heap[Literal(0)].value[0]
|
||||
|
||||
self.heap[Literal(0)] = self.heap[self.size - Literal(1)]
|
||||
self.size -= Literal(1)
|
||||
|
||||
self.max_heapify(Literal(0))
|
||||
self.maxHeapify(Literal(0))
|
||||
|
||||
return max_item
|
||||
|
||||
def peek(self):
|
||||
if self.is_empty():
|
||||
if self.isEmpty():
|
||||
raise IndexError("Queue is empty")
|
||||
return self.heap[Literal(0)]
|
||||
return self.heap[Literal(0)].value[0]
|
||||
|
||||
def is_empty(self):
|
||||
def isEmpty(self):
|
||||
return self.size == Literal(0)
|
||||
|
||||
def __len__(self):
|
||||
"""Gibt die Anzahl der Elemente in der Warteschlange zurück."""
|
||||
return self.size
|
||||
|
||||
def dummify(method):
|
||||
match method.lower():
|
||||
case "add":
|
||||
tmp = MemoryCell(1) + Literal(1);
|
||||
case "cmp":
|
||||
tmp = MemoryCell(1) > Literal(1);
|
||||
|
||||
# Had to adapt different Approach
|
||||
# Had to resort to using primite Types as I do not know the language good enough to get my original
|
||||
# approach working and ran out of time
|
||||
class PriorityQueue_Primitive:
|
||||
def __init__(self, max_size=100):
|
||||
self.heap = [None] * max_size
|
||||
self.size = 0
|
||||
|
||||
def parent(self, i):
|
||||
return (i - 1) // 2
|
||||
|
||||
def leftChild(self, i):
|
||||
return 2 * i + 1
|
||||
|
||||
def rightChild(self, i):
|
||||
return 2 * i + 2
|
||||
|
||||
def swap(self, i, j):
|
||||
self.heap[i], self.heap[j] = self.heap[j], self.heap[i]
|
||||
|
||||
def maxHeapify(self, i):
|
||||
left = self.leftChild(i)
|
||||
right = self.rightChild(i)
|
||||
largest = i
|
||||
|
||||
if left < self.size and self.heap[left] is not None:
|
||||
if self.heap[largest] is None or self.heap[left] > self.heap[largest]:
|
||||
largest = left
|
||||
|
||||
if right < self.size and self.heap[right] is not None:
|
||||
if self.heap[largest] is None or self.heap[right] > self.heap[largest]:
|
||||
largest = right
|
||||
|
||||
if largest != i:
|
||||
self.swap(i, largest)
|
||||
self.maxHeapify(largest)
|
||||
|
||||
def insert(self, entry):
|
||||
if self.size >= len(self.heap):
|
||||
raise IndexError("Heap full")
|
||||
|
||||
i = self.size
|
||||
self.heap[i] = entry
|
||||
self.size += 1
|
||||
|
||||
while i > 0:
|
||||
parent = self.parent(i)
|
||||
|
||||
# Overloaded -> LSP not working quite right, but Interpreter does
|
||||
if self.heap[parent] is None or self.heap[i] > self.heap[parent]:
|
||||
self.swap(i, parent)
|
||||
i = parent
|
||||
else:
|
||||
break
|
||||
|
||||
def pop(self):
|
||||
if self.is_empty():
|
||||
raise IndexError("Queue is empty!")
|
||||
|
||||
max_entry = self.heap[0]
|
||||
|
||||
self.heap[0] = self.heap[self.size - 1]
|
||||
self.heap[self.size - 1] = None
|
||||
self.size -= 1
|
||||
|
||||
if not self.is_empty():
|
||||
self.maxHeapify(0)
|
||||
|
||||
return max_entry
|
||||
|
||||
def peek(self):
|
||||
if self.is_empty():
|
||||
raise IndexError("Queue is empty")
|
||||
return self.heap[0]
|
||||
|
||||
def is_empty(self):
|
||||
return self.size == 0
|
||||
|
||||
def __len__(self):
|
||||
return self.size
|
||||
|
||||
if __name__ == '__main__':
|
||||
pq = PriorityQueue_Primitive()
|
||||
assert(pq.size == 0 and pq.is_empty())
|
||||
test = HeapEntry("A", 2)
|
||||
print(test)
|
||||
a = MemoryArray([test])
|
||||
print(a)
|
||||
print(a[Literal(0)])
|
||||
pq.insert(HeapEntry("D", 3));
|
||||
pq.insert(HeapEntry("A", 4));
|
||||
pq.insert(HeapEntry("T", 2));
|
||||
pq = PriorityQueue()
|
||||
try:
|
||||
pq.pop()
|
||||
assert False, "Queue should be empty"
|
||||
except IndexError:
|
||||
pass
|
||||
assert(pq.isEmpty() and pq.size == Literal(0))
|
||||
entry = HeapEntry("A", 1)
|
||||
pq.insert(entry)
|
||||
assert(not pq.isEmpty() and pq.size == Literal(1))
|
||||
pq.peek()
|
||||
assert(not pq.isEmpty())
|
||||
assert(pq.pop() == HeapEntry("A", 1))
|
||||
assert(pq.isEmpty())
|
||||
pq.insert(HeapEntry("A", 1))
|
||||
pq.insert(HeapEntry("C", 3))
|
||||
pq.insert(HeapEntry("B", 2))
|
||||
assert(pq.size == Literal(3))
|
||||
assert(pq.pop() == HeapEntry("A", 1))
|
||||
assert(pq.pop() == HeapEntry("B", 2))
|
||||
assert(pq.pop() == HeapEntry("C", 3))
|
||||
pq.insert(HeapEntry("A", 1))
|
||||
pq.insert(HeapEntry("C", 3))
|
||||
pq.insert(HeapEntry("B", 2))
|
||||
print(pq.pop())
|
||||
print(pq.pop())
|
||||
print(pq.pop())
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user