Compare commits

...

23 Commits

Author SHA1 Message Date
02d9557e27 Changed formatting of Graphviz Graph 2025-04-26 19:04:37 +02:00
58f66fc1ff Implemented AVL Insert, sort and some debugstuff to find weird pythonissues 2025-04-26 18:54:00 +02:00
552f226e76 Implemented graphvizify 2025-04-24 15:50:39 +02:00
141ff08b82 merge upstream 2025-04-24 12:33:06 +00:00
c382641234 merge upstream 2025-04-23 09:17:26 +00:00
b12e39952d Removed no longer needed function, cleaned up structure 2025-04-23 10:45:53 +02:00
d8a9b29a69 merge upstream 2025-04-23 08:25:29 +00:00
f56b8c7e7a Added extended testsequencies for PriorityQueue 2025-04-23 10:24:39 +02:00
2735abfe81 Removed unneccessary inheritance due to syntax 2025-04-23 10:23:55 +02:00
82fbfa2772 Implemented pr04 (BST) using RAM 2025-04-18 17:18:37 +02:00
b3d3551994 merge upstream 2025-04-18 12:36:50 +00:00
38c099a94e Fixed order of operations in Insert, added serialization and more unittests 2025-04-13 19:04:22 +02:00
2af96a1b4e Cleaned up pr03 2025-04-13 19:03:06 +02:00
2bcd77f9ec Squash merge be/pr03 into main 2025-04-13 13:55:51 +02:00
1b0f9f8c50 merge upstream 2025-04-09 08:24:04 +00:00
f02669d601 Squash merge be/pr02 into main 2025-04-06 14:46:25 +02:00
e19262e818 merge upstream 2025-04-02 09:19:29 +00:00
3926d8d0c7 fixed one-off error and improved call-logic 2025-04-02 11:19:17 +02:00
364590c563 Added and fixed Comments 2025-03-31 14:56:13 +02:00
79f0fc36fd Implemented Algo_4 with O(n) 2025-03-27 16:41:26 +01:00
c0d376cd5c Implemented Algo_3 (n_log(n)) 2025-03-27 16:33:08 +01:00
8df24e2aa1 Copied basic structure, added sanityChecks and implemented Algo_2 2025-03-27 15:25:43 +01:00
bff98d35a7 Added python venv configuration and froze pip-packages of working setup 2025-03-26 22:25:08 +01:00
8 changed files with 1249 additions and 0 deletions

24
activateEnv.srcme Executable file
View File

@ -0,0 +1,24 @@
#!/bin/bash -u
#
# Source python venv (installed via curl https://pyenv.run | bash) to use specific python-version for this project without affecting system
# For Algorithms and Datastructures Class
#
# For WSLg support for matplotlib export="DISPLAY:0" and
# echo "backend: TkAgg" > ~/.config/matplotlib/matplotlibrc
export PYENV_ROOT="$HOME/.pyenv"
export PATH="$PYENV_ROOT/bin:$PATH"
export PYTHONPATH=".:$PYTHONPATH"
eval "$(pyenv init --path)"
eval "$(pyenv init -)"
eval "$(pyenv virtualenv-init -)"
# Create virtualenv if it doesn't exist
if ! pyenv versions | grep -q AUD; then
echo "Creating Python environment..."
pyenv install -s 3.12.0
pyenv virtualenv 3.12.0 AUD
fi
pyenv activate AUD

13
myreqs.txt Normal file
View File

@ -0,0 +1,13 @@
contourpy==1.3.1
cycler==0.12.1
fonttools==4.56.0
kiwisolver==1.4.8
matplotlib==3.10.1
numpy==2.2.4
packaging==24.2
pillow==11.1.0
pygame==2.6.1
pyparsing==3.2.3
python-dateutil==2.9.0.post0
six==1.17.0
tk==0.1.0

165
schoeffelbe/pr01.py Normal file
View File

@ -0,0 +1,165 @@
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.memory_range import mrange
def max_sequence_1(z: MemoryArray):
n = z.length()
m = MemoryCell(MIN_VALUE)
s = MemoryCell()
l = MemoryCell()
r = MemoryCell()
for i in mrange(n):
for j in mrange(i, n):
s.set(0)
for k in mrange(i, j):
s += z[k]
if s > m:
m.set(s)
l.set(i)
r.set(j)
return m, l, r
def max_sequence_2(z: MemoryArray):
n = z.length()
m = MemoryCell(MIN_VALUE)
s = MemoryCell()
l = MemoryCell()
r = MemoryCell()
for i in mrange(n):
s.set(0)
for j in mrange(i, n):
s += z[j]
if s > m:
m.set(s)
l.set(i)
r.set(j)
return m, l, r
def _max_sequence_3_sub(z: MemoryArray, l: Literal, m: Literal, r: Literal):
# find max-sum from Middle to left
linksMax = MemoryCell(MIN_VALUE)
sum = MemoryCell(0)
links = MemoryCell(l)
rechts = MemoryCell(l)
for i in mrange(m, MemoryCell(l)-Literal(1), -1):
sum += z[i]
if sum > linksMax :
linksMax.set(sum)
links.set(i)
# find max-sum from Middle to right
rechtsMax = MemoryCell(MIN_VALUE)
sum.set(0);
# MRange is exclusive
startRight = MemoryCell(1) + m
for i in mrange(startRight, MemoryCell(1) + r):
sum += z[i]
if sum > rechtsMax:
rechtsMax.set(sum)
rechts.set(i)
return (linksMax + rechtsMax), links, rechts
def _max_sequence_3(z: MemoryArray, l: Literal, r: Literal):
# Calc-Vars -> illegal to use Literal(0) here? Probably
# CAN ALLLL BE LITERALS
linksMax = MemoryCell()
linksL = MemoryCell()
linksR = MemoryCell()
rechtsMax = MemoryCell()
rechtsL = MemoryCell()
rechtsR = MemoryCell()
zwiMax = MemoryCell()
zwiL = MemoryCell()
zwiR = MemoryCell()
# Middle
m = MemoryCell()
# Rec-Term - Reached subarray of size 1
if l == r:
return (z[l], l, r)
# calc middle
m.set(MemoryCell(l) + r)
# Use cutoff/floor here, did not check
m //= Literal(2);
# get maxLeft, then maxRight and then cross them (rec)
(linksMax, linksL, linksR) = _max_sequence_3(z, l, m)
startRight = MemoryCell(1) + m
(rechtsMax, rechtsL, rechtsR) = _max_sequence_3(z, startRight, r)
(zwiMax, zwiL, zwiR) = _max_sequence_3_sub(z, l, m, r)
if linksMax >= rechtsMax and linksMax >= zwiMax:
return (linksMax, linksL, linksR)
if rechtsMax >= linksMax and rechtsMax >= zwiMax:
return (rechtsMax, rechtsL, rechtsR)
return (zwiMax, zwiL, zwiR)
# Wrapper for Seq DivAndConquer to keep call/teststructure possible
def max_sequence_3(z: MemoryArray):
# Start with full range
lstart = Literal(0)
rend = Literal(len(z) - 1)
return _max_sequence_3(z, lstart, rend)
def max_sequence_4(z: MemoryArray):
n = z.length()
max = MemoryCell(MIN_VALUE)
aktLinks = MemoryCell()
links = MemoryCell()
rechts = MemoryCell()
aktSum = MemoryCell()
for i in mrange(n):
aktSum += z[i]
if aktSum > max:
max.set(aktSum)
links.set(aktLinks)
rechts.set(i)
# if negative we start new Sum -> Restart must be better than continue
if aktSum < Literal(0):
aktSum.set(0)
aktLinks.set(MemoryCell(1) + i)
return (max, links, rechts)
def example(max_sequence_func):
l = [-59, 52, 46, 14, -50, 58, -87, -77, 34, 15]
print(l)
z = MemoryArray(l)
m, l, r = max_sequence_func(z)
print(m, l, r)
assert(m == Literal(120))
def seq(filename, max_sequence_func):
z = MemoryArray.create_array_from_file(filename)
m, l, r = max_sequence_func(z)
print(m, l, r)
def analyze_complexity(max_sequence_func, sizes):
"""
Analysiert die Komplexität einer maximalen Teilfolgenfunktion.
:param max_sequence_func: Die Funktion, die analysiert wird.
:param sizes: Eine Liste von Eingabegrößen für die Analyse.
"""
for size in sizes:
MemoryManager.purge() # Speicher zurücksetzen
random_array = MemoryArray.create_random_array(size, -100, 100)
max_sequence_func(random_array)
MemoryManager.save_stats(size)
MemoryManager.plot_stats(["cells", "adds"])
if __name__ == '__main__':
# fn = max_sequence_4
for fn in [max_sequence_1, max_sequence_2, max_sequence_3, max_sequence_4]:
example(fn)
# for filename in ["data/seq0.txt", "data/seq1.txt", "data/seq2.txt", "data/seq3.txt"]:
for filename in ["data/seq0.txt", "data/seq1.txt"]:
print(filename)
seq(filename, fn)
analyze_complexity(fn, [10, 20, 30, 40, 50, 60, 70, 80, 90, 100])

125
schoeffelbe/pr02.py Normal file
View File

@ -0,0 +1,125 @@
import logging
logger = logging.getLogger(__name__)
# logging.basicConfig(level=logging.DEBUG)
import time
def timeMS(func, *args, **kwargs):
startTime = time.perf_counter()
result = func(*args, **kwargs)
endTime = time.perf_counter()
elapsedMS = (endTime - startTime) * 1000 # Convert to milliseconds
print(f"{func.__name__} took {elapsedMS:.2f} ms")
return result
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.memory_range import mrange
def example():
initial = [6, 5, 3, 8, 1, 7, 2, 4]
# initial = [-6, -5, -3, -8, 1, 7, 2, 4]
toSort = MemoryArray(initial)
# init_from_size not accessible?
sorted = MemoryArray([-1] * len(initial))
mergeSort(toSort, sorted)
logger.debug(f"sorted {sorted} vs initial {initial}")
assert all(sorted[Literal(i)] == Literal(i+1) for i in range(len(initial))), "Array not sorted correctly"
analyze_complexity(mergeSort, [10, 20, 30, 40, 50, 60, 70, 80, 90, 100])
def merge(left: MemoryArray, right: MemoryArray, sort: MemoryArray):
pointerLeft = MemoryCell(0)
pointerRight = MemoryCell(0)
pointerSort = MemoryCell(0)
compare = lambda x, y: x <= y
logger.debug(f"Merging left {left} with right {right} in sort {sort}")
while pointerLeft < left.length() and pointerRight < right.length():
if compare(left[pointerLeft], right[pointerRight]):
sort[pointerSort] = left[pointerLeft]
pointerLeft += Literal(1)
else:
sort[pointerSort] = right[pointerRight]
pointerRight += Literal(1)
logger.debug(f"Now are at sort {sort} with {pointerLeft} (l) and {pointerRight} (r)")
pointerSort += Literal(1)
# Consume remaining elements
while pointerLeft < left.length():
logger.debug(f"Consuming left {left} from {pointerSort} at {pointerLeft}")
sort[pointerSort] = left[pointerLeft]
pointerLeft += Literal(1)
pointerSort += Literal(1)
while pointerRight < right.length():
logger.debug(f"Consuming right {right} from {pointerSort} at {pointerRight}")
sort[pointerSort] = right[pointerRight]
pointerRight += Literal(1)
pointerSort += Literal(1)
# Sort the array passed as "toSort" and place the result in array "sort"
# Does not change the original Array
def mergeSort(toSort: MemoryArray, sort: MemoryArray):
logger.debug(toSort)
toSortLength = MemoryCell(toSort.length())
# Splitting
# Rec-Term -> Reached single Element. Single Element is already sorted so we place it!
if toSortLength <= Literal(1):
# still working for empty array
if toSortLength == Literal(1):
sort[Literal(0)] = toSort[Literal(0)]
return
# TODO - Use a global var or a reference to an array passed as argument for this
# TODO - Tried non-temp-array approach with alternating Work-Arrays passed to the function, but made code really unreadable. Decided not worth it for now
# Temporary Arrays to hold the split arrays
mid : Literal = toSortLength // Literal(2)
left : MemoryArray = MemoryArray([toSort[i] for i in mrange(mid)])
right : MemoryArray = MemoryArray([toSort[i] for i in mrange(mid, toSortLength)])
# Temporary arrays for sorted halves
leftSort = MemoryArray([-1] * mid.get())
rightSort = MemoryArray([-1] * (toSortLength - mid).get())
# Split further
mergeSort(left, leftSort)
mergeSort(right, rightSort)
# Recreate the array from the seperated parts
merge(leftSort, rightSort, sort)
def analyze_complexity(fn, sizes):
"""
Analysiert die Komplexität einer maximalen Teilfolgenfunktion.
:param max_sequence_func: Die Funktion, die analysiert wird.
:param sizes: Eine Liste von Eingabegrößen für die Analyse.
"""
for size in sizes:
MemoryManager.purge() # Speicher zurücksetzen
random_array = MemoryArray.create_random_array(size, -100, 100)
other_array = MemoryArray([-1] * size)
fn(random_array, other_array)
MemoryManager.save_stats(size)
MemoryManager.plot_stats(["cells", "adds", "compares"])
if __name__ == '__main__':
# For debug, assert if working and complexity-analysis
# example()
for filename in ["data/seq0.txt", "data/seq1.txt", "data/seq2.txt", "data/seq3.txt"]:
print(filename)
toSort = MemoryArray.create_array_from_file(filename)
sorted = MemoryArray([-1] * toSort.length().get())
timeMS(mergeSort, toSort, sorted)
# print(sorted)

190
schoeffelbe/pr03.py Normal file
View File

@ -0,0 +1,190 @@
import logging
logger = logging.getLogger(__name__)
# logging.basicConfig(level=logging.DEBUG)
import time
def timeMS(func, *args, **kwargs):
startTime = time.perf_counter()
result = func(*args, **kwargs)
endTime = time.perf_counter()
elapsedMS = (endTime - startTime) * 1000 # Convert to milliseconds
print(f"{func.__name__} took {elapsedMS:.2f} ms")
return result
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.memory_range import mrange
def example():
initial = [6, 5, 3, 8, 1, 7, 2, 4]
# initial = [-6, -5, -3, -8, 1, 7, 2, 4]
toSort = MemoryArray(initial)
quickSortIterative(toSort, Literal(0), toSort.length().pred())
logger.debug(f"sorted {toSort} vs initial {initial}")
assert all(toSort[Literal(i)] == Literal(i+1) for i in range(len(initial))), "Array not sorted correctly"
# analyze_complexity(quickSort, [10, 20, 30, 40, 50, 60, 70, 80, 90, 100])
def getPivot(z: MemoryArray, l: Literal, r: Literal, mode) -> Literal:
if mode == 0:
return r
else:
mid_offset = r.value - l.value
mid_offset = mid_offset // 2
mid = Literal(l.value + mid_offset)
# Return median of left, middle, and right elements
if ((z[l] <= z[mid] and z[mid] <= z[r]) or
(z[r] <= z[mid] and z[mid] <= z[l])):
return mid
elif ((z[mid] <= z[l] and z[l] <= z[r]) or
(z[r] <= z[l] and z[l] <= z[mid])):
return l
else:
return r
def swap(z: MemoryArray, i: int, j: int):
tmp = z[Literal(i)].value
z[Literal(i)] = z[Literal(j)]
z[Literal(j)].set(tmp)
# toSort[] --> Array to be sorted,
# left --> Starting index,
# right --> Ending index
# adapted from https://stackoverflow.com/questions/68524038/is-there-a-python-implementation-of-quicksort-without-recursion
def quickSortIterative(toSort : MemoryArray, left : Literal, right : Literal, mode=0):
# Create a manually managed stack and avoid pythons recursion-limit
size = right.value - left.value + 1
stack : MemoryArray = MemoryArray([0] * size)
top : MemoryCell = MemoryCell(-1)
# push initial values of l and h to stack
top += Literal(1)
stack[top] = left
top += Literal(1)
stack[top] = right
# Keep popping from stack until its empty
while top >= Literal(0):
logger.debug(f"size {size}, stack {stack}, right {right} and left {left}, top {top}")
# Pop h and l - Ensure we are not getting them by Ref, this will produce weird "JUST A LITTLE OF" Results
right = Literal(stack[top].get())
top -= Literal(1)
left = Literal(stack[top].get())
top -= Literal(1)
# Set pivot element at its correct position in sorted array
p = partitionIterative(toSort, left, right, mode)
# If there are elements on left side of pivot, then push left side to stack
if p.pred() > left:
top += Literal(1)
stack[top] = left
top += Literal(1)
stack[top] = p.pred()
# If there are elements on right side of pivot, then push right side to stack
if p.succ() < right:
top += Literal(1)
stack[top] = p.succ()
top += Literal(1)
stack[top] = right
def partitionIterative(arr : MemoryArray, l : Literal, h : Literal, mode=0):
logger.debug(f"Partitioning {arr}, {l} and {h}")
pivot_idx : Literal = getPivot(arr, l, h, mode)
# If pivot isn't at the high end, swap it there
if pivot_idx != h:
swap(arr, int(pivot_idx), int(h))
# Carefull that we do not use a reference. I suppose python would return one here if we just assign without value>Literal cast.
# At least this helped fix weird issue
pivotValue : Literal = arr[h]
i : MemoryCell = MemoryCell(l.pred())
for j in mrange(l, h):
if arr[j] <= pivotValue:
i += Literal(1) # increment index of smaller element
swap(arr, int(i), int(j))
swap(arr, i.succ().value, h.value)
return i.succ()
def LEGACY_quickSort(z: MemoryArray, l: Literal = Literal(0), r: Literal = Literal(-1), mode=0):
if r == Literal(-1):
r = z.length().pred();
if l < r:
q = LEGACY_partition(z, l, r, mode)
LEGACY_quickSort(z, l, q.pred())
LEGACY_quickSort(z, q.succ(), r)
def LEGACY_partition(z: MemoryArray, l: Literal, r: Literal, mode):
# Get pivot
pivot_idx = getPivot(z, l, r, mode)
# If pivot is not already at the right end, swap it there
if pivot_idx != r:
swap(z, int(pivot_idx), int(r))
with MemoryCell(z[r]) as pivot, MemoryCell(l) as i, MemoryCell(r.pred()) as j:
while i < j:
while z[i] < pivot:
i.set(i.succ())
while j > l and z[j] >= pivot:
j.set(j.pred())
if i < j:
swap(z, int(i), int(j))
i.set(i.succ())
j.set(j.pred())
if i == j and z[i] < pivot:
i.set(i.succ())
if z[i] != pivot:
swap(z, int(i), int(r))
return Literal(i)
def analyze_complexity(fn, sizes):
"""
Analysiert die Komplexität einer maximalen Teilfolgenfunktion.
:param max_sequence_func: Die Funktion, die analysiert wird.
:param sizes: Eine Liste von Eingabegrößen für die Analyse.
"""
for size in sizes:
MemoryManager.purge() # Speicher zurücksetzen
random_array = MemoryArray.create_random_array(size, -100, 100)
fn(random_array, Literal(0), random_array.length().pred())
MemoryManager.save_stats(size)
MemoryManager.plot_stats(["cells", "adds", "compares", "reads", "writes"])
if __name__ == '__main__':
# For debug, assert if working and complexity-analysis
example()
print("I ran into a MaxRecursionDepth Error. From what I read on the Internet python does not do Tailcall Optimizations")
print("Increasing recursion-limit seems like a poor Idea, therefore tried an iterative approach with manual stack-keeping")
toSort = MemoryArray.create_array_from_file("data/seq0.txt")
print(toSort)
quickSortIterative(toSort, Literal(0), toSort.length().pred())
print(toSort)
# analyze_complexity(quickSortIterative, [10, 20, 30, 40, 50, 60, 70, 80, 90, 100])
for filename in ["data/seq0.txt", "data/seq1.txt", "data/seq2.txt" ,"data/seq3.txt"]:
# for filename in [ "data/seq1.txt"]:
print(filename)
toSort = MemoryArray.create_array_from_file(filename)
timeMS(quickSortIterative, toSort, Literal(0), toSort.length().pred(), mode=1)
print(toSort)
print("Kann durch die Modifikation eine besser Laufzeit als nlog(n) erreicht werden? Nein! nlog(n) ist das Minimum. Durch die Änderung kann aber der Worst-Case fall von n^2 für z.B. bereits vorsortierte Arrays oder Arrays mit vielen Duplikaten vermieden werden.")

293
schoeffelbe/pr04.py Normal file
View File

@ -0,0 +1,293 @@
import logging
logger = logging.getLogger(__name__)
# logging.basicConfig(level=logging.DEBUG)
import time
def timeMS(func, *args, **kwargs):
startTime = time.perf_counter()
result = func(*args, **kwargs)
endTime = time.perf_counter()
elapsedMS = (endTime - startTime) * 1000 # Convert to milliseconds
print(f"{func.__name__} took {elapsedMS:.2f} ms")
return result
from utils.memory_array import MemoryArray
from utils.memory_cell import MemoryCell
from utils.literal import Literal
from utils.constants import MAX_VALUE
from utils.memory_manager import MemoryManager
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))
# 0: Content, 1: Prio
self.data[Literal(0)] = Literal(item)
self.data[Literal(1)] = Literal(priority)
def getItem(self):
return self.data[Literal(0)]
def getPriority(self):
return self.data[Literal(1)]
def setPriority(self, priority):
self.data[Literal(1)] = Literal(priority)
def __lt__(self, other):
if other is None:
return True
if isinstance(other, (int, float)):
return self.getPriority().value > other
return self.getPriority() > other.getPriority()
def __gt__(self, other):
if other is None:
return False
if isinstance(other, (int, float)):
return self.getPriority().value < other
return self.getPriority() < other.getPriority()
def __eq__(self, other):
return self.getPriority() == other.getPriority()
def __str__(self):
return f"({self.getItem()}, prio={self.getPriority()})"
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:
return MemoryCell(i.pred()) // Literal(2)
def leftChild(self, i: Literal) -> Literal:
return MemoryCell(MemoryCell(2) * i) + Literal(1)
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 maxHeapify(self, i: Literal):
left = self.leftChild(i)
right = self.rightChild(i)
largest = i
if left < Literal(self.size.value) and self.heap[left].value[0] > self.heap[largest].value[0]:
largest = left
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.maxHeapify(largest)
def insert(self, entry : HeapEntry):
if self.size >= self.heap.length():
raise IndexError("Heap full")
i = self.size
self.heap[i].set([entry])
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)
self.size += Literal(1)
def pop(self):
if self.isEmpty():
raise IndexError("Queue is empty!")
max_item = self.heap[Literal(0)].value[0]
self.heap[Literal(0)] = self.heap[self.size - Literal(1)]
self.size -= Literal(1)
self.maxHeapify(Literal(0))
return max_item
def peek(self):
if self.isEmpty():
raise IndexError("Queue is empty")
return self.heap[Literal(0)].value[0]
def isEmpty(self):
return self.size == Literal(0)
def __len__(self):
return self.size
def __str__(self):
entries = []
for i in mrange(self.size.value):
entry = self.heap[i].value[0]
if entry.getItem() != MAX_VALUE:
entries.append(str(entry))
return "[" + ", ".join(entries) + "]"
# Insert here so we dont run into import problems, but can deliver this file Standalone
class BinaryTreeNode(MemoryCell):
def __init__(self, value):
super().__init__(value)
self.left = None
self.right = None
def __repr__(self):
return f"BinaryTreeNode(value={self.value}, left={self.left}, right={self.right})"
def __str__(self):
return str(self.value)
class BinaryTree:
def __init__(self):
self.root: BinaryTreeNode | None = None
def insert(self, value: BinaryTreeNode):
# Insert at Leaf, if smaller then left one, otherwise right one
def _insert(node: BinaryTreeNode | None, value) -> BinaryTreeNode:
if node is None:
return BinaryTreeNode(value)
if value < node:
node.left = _insert(node.left, value) # type: ignore -> Ignoring pywright errors
else:
node.right = _insert(node.right, value) # type: ignore -> Ignoring pywright errors
return node
self.root = _insert(self.root, value)
def traverse(self, mode="in", visual=False):
mode = mode.lower()
# Have internal depth counting
def InternalTraverse(node, prefix="", is_left=True, depth=0):
if node is None:
return [] if not visual else []
result = []
node_str = str(node)
prefixAcc = prefix + ("| " if is_left and depth > 0 else " ")
if visual:
connector = "+-- " if is_left else "L-- "
line = prefix + connector + node_str if depth > 0 else node_str
result.append(line)
else:
result.append(node_str)
if mode == "pre":
result += InternalTraverse(node.left, prefixAcc, True, depth + 1)
result += InternalTraverse(node.right, prefixAcc, False, depth + 1)
elif mode == "in":
result += InternalTraverse(node.left, prefixAcc, True, depth + 1)
result += InternalTraverse(node.right, prefixAcc, False, depth + 1)
elif mode == "post":
result += InternalTraverse(node.left, prefixAcc, True, depth + 1)
result += InternalTraverse(node.right, prefixAcc, False, depth + 1)
return result
if self.root is None:
return "(empty tree)" if visual else []
result = InternalTraverse(self.root)
return "\n".join(result) if visual else result
def levelOrderWithPriorityQueue(self):
if not self.root:
return []
# Create a priority queue, using a reduced prio for every new entry -> behaviour as regular queue FIFO
pq = PriorityQueue(Literal(1000))
# Again we cannot create a MemoryArray of dynamic sizes and also cannot create a string as MemoryCell does not like it
# Again we just create a list holding a single dummy Entry (to set its size to 1) and then just use this "list" as our string
# Appending to it is easy as it is just a regular list and in the end we return it
# Like MemoryCell("").value.append("STRING") will fail. But list-wrap works.
#
# Sorry for Syntax, dont know any better way to have everything as RAM-Managed memory:-(
result = MemoryArray(["MYSTRING"])
result[Literal(0)].set([]);
counter = MemoryCell(0)
def nextPriority():
val = counter.value
counter.set(Literal(val + 1))
return val
pq.insert(HeapEntry([self.root], nextPriority()))
while not pq.isEmpty():
entry = pq.pop()
node = entry.getItem().value
result[Literal(0)].value.append(str(node[0]))
if node[0].left:
pq.insert(HeapEntry([node[0].left], nextPriority()))
if node[0].right:
pq.insert(HeapEntry([node[0].right], nextPriority()))
return result[Literal(0)]
def __str__(self):
return str(self.traverse(mode="PrE", visual=True))
def analyze_complexity(fn, sizes):
"""
Analysiert die Komplexität einer maximalen Teilfolgenfunktion.
:param max_sequence_func: Die Funktion, die analysiert wird.
:param sizes: Eine Liste von Eingabegrößen für die Analyse.
"""
for size in sizes:
MemoryManager.purge() # Speicher zurücksetzen
random_array = MemoryArray.create_random_array(size, -100, 100)
fn(random_array, Literal(0), random_array.length().pred())
MemoryManager.save_stats(size)
MemoryManager.plot_stats(["cells", "adds", "compares", "reads", "writes"])
if __name__ == '__main__':
# For debug, assert if working and complexity-analysis
# example()
print("Sorry for the Syntax and the large file, tried to keep everything as a standalone file to help make it \" download and run \".\n \
Also did - once again - not find a better way to have a queue managed by the RAM contain the values of non-integer-attributes I \n\
needed it to. Therefore i reused my Priorityqueue and its accesses via the unspecified wrapped list.");
# for filename in ["data/seq0.txt", "data/seq1.txt", "data/seq2.txt" ,"data/seq3.txt"]:
for filename in [ "data/seq0.txt"]:
print(filename)
binTreeData = MemoryArray.create_array_from_file(filename)
binTree = BinaryTree()
for value in binTreeData:
binTree.insert(BinaryTreeNode(value))
# Print overlaoded InOrder traversal
print(binTree)
# print(binTree.traverse(mode="pre", visual=False))
# print(binTree.traverse(mode="in", visual=False))
# print(binTree.traverse(mode="post", visual=False))
# Print Levelorder traversal:
print(binTree.levelOrderWithPriorityQueue())

246
schoeffelbe/pr05.py Normal file
View File

@ -0,0 +1,246 @@
import logging
logger = logging.getLogger(__name__)
# logging.basicConfig(level=logging.DEBUG)
import time
def timeMS(func, *args, **kwargs):
startTime = time.perf_counter()
result = func(*args, **kwargs)
endTime = time.perf_counter()
elapsedMS = (endTime - startTime) * 1000 # Convert to milliseconds
print(f"{func.__name__} took {elapsedMS:.2f} ms")
return result
from utils.memory_array import MemoryArray
from utils.literal import Literal
from utils.memory_manager import MemoryManager
from vorlesung.L05_binaere_baeume.bin_tree import BinaryTree
from vorlesung.L05_binaere_baeume.bin_tree_node import BinaryTreeNode
def analyze_complexity(fn, sizes):
"""
Analysiert die Komplexität einer maximalen Teilfolgenfunktion.
:param max_sequence_func: Die Funktion, die analysiert wird.
:param sizes: Eine Liste von Eingabegrößen für die Analyse.
"""
for size in sizes:
MemoryManager.purge() # Speicher zurücksetzen
random_array = MemoryArray.create_random_array(size, -100, 100)
fn(random_array, Literal(0), random_array.length().pred())
MemoryManager.save_stats(size)
MemoryManager.plot_stats(["cells", "adds", "compares", "reads", "writes"])
lineAccumulator = []
# Returnvalue does not get forwarded so we can not work with return.
# Will try glob vars to append the string
# Signature: def print_node(node, indent=0, line=None):
def clbk_graphvizify(toDecorate : BinaryTreeNode, indent=0, line=None):
global lineAccumulator
if isinstance(toDecorate, AVLTreeNode):
lineAccumulator.append(f'n_{id(toDecorate)} [label=<{toDecorate.value}<BR/><FONT COLOR="RED" POINT-SIZE="10.0" FACE="ambrosia">B: {toDecorate.balanceFactor}</FONT>>]')
else:
lineAccumulator.append(f"n_{id(toDecorate)} [label={toDecorate.value}]")
# Create edges for nodes with Child (use l - r)
if toDecorate.left is not None:
lineAccumulator.append(f"n_{id(toDecorate)} -> n_{id(toDecorate.left)}")
if toDecorate.right is not None:
lineAccumulator.append(f"n_{id(toDecorate)} -> n_{id(toDecorate.right)}")
def graphvizify() -> str:
# Header
result = "digraph {\n\t"
# Body
result += ('\n\t'.join(str(item) for item in lineAccumulator))
# Footer
result += "\n}"
return result
class AVLTreeNode(BinaryTreeNode):
def __init__(self, value):
super().__init__(value)
self.parentRef = None
# Start balanced as we probably have no children right after insert
self.balanceFactor = Literal(0)
def rightRotate(self, node) -> 'AVLTreeNode|None':
if node is None:
return None
oLeft = node.left;
oLeft.parentRef = node.parentRef;
node.left = oLeft.right;
if node.left is not None:
node.left.parentRef = node;
oLeft.right = node;
node.parentRef = oLeft;
if oLeft.parentRef is not None:
if oLeft.parentRef.right is node:
oLeft.parentRef.right = oLeft;
elif oLeft.parentRef.left is node:
oLeft.parentRef.left = oLeft;
node.getSetBalanceFactor()
oLeft.getSetBalanceFactor()
return oLeft
def leftRotate(self, node) -> 'AVLTreeNode|None':
if node is None:
return None
oRight = node.right
oRight.parentRef = node.parentRef
node.right = oRight.left
if node.right is not None:
node.right.parentRef = node
oRight.left = node
node.parentRef = oRight
if oRight.parentRef is not None:
if oRight.parentRef.right is node:
oRight.parentRef.right = oRight
elif oRight.parentRef.left is node:
oRight.parentRef.left = oRight
node.getSetBalanceFactor()
oRight.getSetBalanceFactor()
return oRight
def getSetBalanceFactor(self) -> Literal:
leftHeight = self.left.height() if self.left else 0
rightHeight = self.right.height() if self.right else 0
self.balanceFactor = Literal(rightHeight - leftHeight)
return self.balanceFactor
def rightLeftRotate(self, node) -> 'AVLTreeNode|None':
node.right = self.rightRotate(node.right)
return self.leftRotate(node)
def leftRightRotate(self, node) -> 'AVLTreeNode|None':
node.left = self.leftRotate(node.left)
return self.rightRotate(node)
def debugTraverse(node, source=1):
if node is None:
return None
logger.debug(f"{node.value} {node.getSetBalanceFactor()} {source}")
debugTraverse(node.left, 10);
debugTraverse(node.right, 20);
class AVLTree(BinaryTree):
# @override
def new_node(self, value) -> AVLTreeNode:
return AVLTreeNode(value)
def balanceAVLTree(self, node : AVLTreeNode):
# balance < -1 means imbalance to the left, > 1 means imbalance to the right
logger.debug("in")
if node is None:
return None
logger.debug("out")
node.getSetBalanceFactor()
logger.debug(f"Parent Balancing for {node.value} -> {node.balanceFactor} {node.left.height() if node.left else None} and {node.right.height() if node.right else None}")
# imbalance to left -> If we enter this we cannot LOGICALLY have a left=None node -> No need to chekc
if node.balanceFactor < Literal(-1):
# Left-Left
if node.left.balanceFactor <= Literal(0): # type: ignore -> Ignoring pywright error, see comment above
# Wow, this syntax is sketchy ^^
# TODO Maybe declare as static if python supports this? Or just leaf param be?
logger.debug("rr")
node = node.rightRotate(node)
# Left-Right
else:
# TODO Maybe declare as static if python supports this? Or just leaf param be?
logger.debug("lrr")
node = node.leftRightRotate(node)
# Right heavy
# imbalance to right -> If we enter this we cannot LOGICALLY have a right=None node -> No need to chekc
if node.balanceFactor > Literal(1):
# Right-Right case
if node.right.balanceFactor >= Literal(0): # type: ignore -> Ignoring pywright error, see comment above
# TODO Maybe declare as static if python supports this? Or just leaf param be?
logger.debug("lr")
node = node.leftRotate(node)
# Right-Left case
else:
# TODO Maybe declare as static if python supports this? Or just leaf param be?
logger.debug("rlr")
node = node.rightLeftRotate(node)
logger.debug(f"Reached {node.parentRef}")
if node.parentRef is not None:
logger.debug(f"Calling again for {node.parentRef.value}");
self.balanceAVLTree(node.parentRef);
else:
self.root = node;
# Node is balanced
return node
# @override
def insert(self, value):
node, parent = super().insert(value)
# NOTE Python does not have a Problem with NOT tellin us that we override something important
# or something that does not exist.... This Makes for AWESOME debugging .... ... ...
node.parentRef = parent
if parent:
node = self.balanceAVLTree(node.parentRef)
return node, parent
if __name__ == '__main__':
tree = AVLTree()
### Force RR
# testData = [30, 20, 10];
# for value in testData:
# tree.insert(MemoryCell(value));
### Force LR
# testData = [10, 20, 30];
# for value in testData:
# tree.insert(MemoryCell(value));
### Force LRR
# testData = [30, 10, 20]
# for value in testData:
# tree.insert(MemoryCell(value))
### Force RLR
# testData = [10, 30, 20]
# for value in testData:
# tree.insert(MemoryCell(value))
# Force rebuild of our balanceFactor indices...
# debugTraverse(tree.root)
binTreeData = MemoryArray.create_array_from_file("data/seq0.txt")
for value in binTreeData:
tree.insert(value)
lineAccumulator.clear();
tree.in_order_traversal(clbk_graphvizify)
# tree.tree_structure_traversal(clbk_graphvizify)
print(graphvizify())

View File

@ -0,0 +1,193 @@
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_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))
# 0: Content, 1: Prio
self.data[Literal(0)] = Literal(item)
self.data[Literal(1)] = Literal(priority)
def getItem(self):
return self.data[Literal(0)]
def getPriority(self):
return self.data[Literal(1)]
def setPriority(self, priority):
self.data[Literal(1)] = Literal(priority)
def __lt__(self, other):
if other is None:
return True
if isinstance(other, (int, float)):
return self.getPriority().value > other
return self.getPriority() < other.getPriority()
def __gt__(self, other):
if other is None:
return False
if isinstance(other, (int, float)):
return self.getPriority().value < other
return self.getPriority() > other.getPriority()
def __eq__(self, other):
return self.getPriority() == other.getPriority()
def __str__(self):
return f"({self.getItem()}, prio={self.getPriority()})"
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(MIN_VALUE, MIN_VALUE)])
self.size = MemoryCell(0)
def parent(self, i: Literal) -> Literal:
return MemoryCell(i.pred()) // Literal(2)
def leftChild(self, i: Literal) -> Literal:
return MemoryCell(MemoryCell(2) * i) + Literal(1)
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 maxHeapify(self, i: Literal):
left = self.leftChild(i)
right = self.rightChild(i)
largest = i
if left < Literal(self.size.value) and self.heap[left].value[0] > self.heap[largest].value[0]:
largest = left
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.maxHeapify(largest)
def insert(self, entry : HeapEntry):
if self.size >= self.heap.length():
raise IndexError("Heap full")
i = self.size
self.heap[i].set([entry])
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)
self.size += Literal(1)
def pop(self):
if self.isEmpty():
raise IndexError("Queue is empty!")
max_item = self.heap[Literal(0)].value[0]
self.heap[Literal(0)] = self.heap[self.size - Literal(1)]
self.size -= Literal(1)
self.maxHeapify(Literal(0))
return max_item
def peek(self):
if self.isEmpty():
raise IndexError("Queue is empty")
return self.heap[Literal(0)].value[0]
def isEmpty(self):
return self.size == Literal(0)
def __len__(self):
return self.size
def __str__(self):
entries = []
for i in mrange(self.size.value):
entry = self.heap[i].value[0]
if entry.getItem() != MIN_VALUE:
entries.append(str(entry))
return "[" + ", ".join(entries) + "]"
def testQueueRandom(number: int):
import random
import string
pq = PriorityQueue(Literal(number))
entries = []
for _ in range(number):
value = ''.join(random.choices(string.ascii_uppercase + string.digits, k=3))
priority = random.randint(1, 100)
entry = HeapEntry(value, priority)
entries.append(entry)
pq.insert(entry)
print(pq)
for entry in entries:
print(f"Unprioritized: {entry}")
while not pq.isEmpty():
print(pq.pop())
if __name__ == '__main__':
# Proof of Concept
testEntry = HeapEntry("A", 2)
print(testEntry)
testArray = MemoryArray([testEntry])
print(testArray)
print(testArray[Literal(0)])
# Queue Testing
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("C", 3))
pq.insert(HeapEntry("B", 2))
pq.insert(HeapEntry("A", 1))
assert(pq.size == Literal(3))
assert(pq.pop() == HeapEntry("C", 3))
assert(pq.pop() == HeapEntry("B", 2))
assert(pq.pop() == HeapEntry("A", 1))
pq.insert(HeapEntry("A", 1))
pq.insert(HeapEntry("C", 3))
pq.insert(HeapEntry("B", 2))
pq.insert(HeapEntry(42, 4))
pq.insert(HeapEntry(42, 1))
pq.insert(HeapEntry("C", 2))
print(pq)
while not pq.isEmpty():
print(pq.pop())
testQueueRandom(100)