Implemented pr04 (BST) using RAM

schoeffelbe/pr04.py

@@ -0,0 +1,312 @@
+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(BinaryTreeNode):
+    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 preOrder(self, node : BinaryTreeNode):
+        if node is None:
+            return []
+        return [str(node)] + self.preOrder(node.left) + self.preOrder(node.right)     # type: ignore -> Ignoring pywright errors
+
+    def inOrder(self, node : BinaryTreeNode):
+        if node is None:
+            return []
+        return self.inOrder(node.left) + [str(node)] + self.inOrder(node.right)       # type: ignore -> Ignoring pywright errors
+
+    def postOrder(self, node : BinaryTreeNode):
+        if node is None:
+            return []
+        return self.postOrder(node.left) + self.postOrder(node.right) + [str(node)]   # type: ignore -> Ignoring pywright errors
+
+    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)
+
+            if mode == "pre":
+                if not visual:
+                    result.append(node_str)
+                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)
+                if not visual:
+                    result.append(node_str)
+                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)
+                if not visual:
+                    result.append(node_str)
+    
+            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())