PriorityQueue Implementation

LinkedQueue.py

""" Implementation of
-> LinkedList
-> Singly Linked List
"""

class Node:
    def __init__(self, data, next=None):
        self.data = data
        self.next = next

class LinkedQueue:

    def __init__(self):
        self.head = None
        self.tail = None
        self.size = 0
        self.array = []

    def __iter__(self):
        if len(self.array) > 0:
            self.array = []
        current = self.head
        while current is not None:
            self.array.append(current.data)
            current = current.next
        return self.array

    def is_empty(self):
        return self.size == 0

    def FirstIn(self, data):
        """ Append operation of LL """
        newNode = Node(data)
        if self.head is None:
            self.head = self.tail = newNode
        else:
            self.tail.next = newNode
            self.tail = newNode
        self.size += 1
        return self

    def FirstOut(self):
        """ list.pop(0) operation of LL """
        if self.head is None:
            raise Exception('LinkedList empty')
        else:
            temp = self.head
            self.head = self.head.next
        self.size -= 1
        return temp.data


class SinglyLinkedList:
    """ Implemeting SLL for priority queue """

    def __init__(self):
        self.head = None
        self.tail = None
        self.size = 0
        self.array = []

    def is_empty(self):
        return self.size == 0

    def __len__(self):
        return self.size

    def __iter__(self):
        if len(self.array) > 0:
            self.array = []
        current = self.head
        while current is not None:
            self.array.append(current.data)
            current = current.next

        return self.array

    def insertHead(self, data):
        newNode = Node(data)
        if self.is_empty():
            self.head = self.tail = newNode
        else:
            newNode.next = self.head
            self.head = newNode
        self.size += 1
        return self

    def insertTail(self, data):
        newNode = Node(data)
        if self.head is None:
            raise Exception('List Empty')
        else:
            self.tail.next = newNode
            self.tail = newNode
        self.size += 1
        return self

    def deleteHead(self):
        assert self.head is not None, 'List Empty'
        temp = self.head
        self.head = self.head.next
        self.size -= 1
        return temp.data

    def deleteTail(self):
        """ Complexity -> O(n) """
        assert self.head is not None, 'List Empty'
        if self.size == 1:
            self.tail = None
        current = self.head
        while current.next is not self.tail:
            current = current.next

        temp = current.next
        self.tail = current
        current.next = None
        return temp.data

priority_queue.py

""" Implementing
-> Priority Queues
   |_ Unbounded PQ
      |_ Arrays
      |_ Linked List
   |_ Bounded PQ

Requirement
-----------
    1. Queue Basics
    2. OOP's

"""

from LinkedList import Node, LinkedQueue

# Base class for defining each item with priority


class PriorityQEntry:
    def __init__(self, item, key):
        self.data = item
        self.priority = key

    def __lt__(self, other_key):
        """ compare current key with other_key """

        return "Highest" if self.priority < other_key else "lowest"


class DllNode:
    def __init__(self, element, prev=None, next=None):
        self.element = element
        self.prev = prev
        self.next = next


class UnboundedPriorityQueueArray:
    """ Unbounded PriorityQueue """

    def __init__(self):
        self.array = list()
        self.size = 0

    def is_empty(self):
        return self.size == 0

    def __len__(self):
        return self.size

    def getHighestPriorityIndex(self):
        min_ = 0
        for i in range(1, len(self.array)):
            if self.array[min_].priority > self.array[i].priority:
                min_ = i

        return min_

    def peek(self):
        """ Return the highest priority element """

        if self.is_empty():
            raise Exception(" Priority Queue Empty")

        index = self.getHighestPriorityIndex()
        return self.array[index].data

    def enqueue(self, item, key):
        """ Simple Enqueue operation
        time- O(1), Space- O(n)
        """

        obj = PriorityQEntry(item, key)
        self.array.append(obj)
        self.size += 1
        return self

    def dequeue(self):
        """ Remove the item by searching the highest priority key
        time- O(n)
        """

        if self.is_empty():
            raise Exception(" Priority Queue Empty")

        index = self.getHighestPriorityIndex()
        temp = self.array.pop(index)
        return temp.data


class UnboundedPriorityQueueDLL:
    def __init__(self):
        self.head = None
        self.tail = None
        self.size = 0
        self.array = []

    def is_empty():
        return self.size == 0

    def __iter__(self):
        if len(self.array) > 0:
            self.array = []
        current = self.head
        while current is not None:
            self.array.append(current.element.data)
            current = current.next
        return self.array

    def enqueue(self, key, item):
        """ Enqueue in the end
        time- O(1) """

        PQobj = DllNode(PriorityQEntry(key, item))
        if self.head is None:
            self.head = self.tail = PQobj
            self.head.next = self.tail
            self.tail.prev = self.head
        else:
            PQobj.prev = self.tail
            self.tail.next = PQobj
            self.tail = self.tail.next
        self.size += 1
        return self

    def dequeue(self):
        """ Search the highest priority element
        time- O(n)
        """

        assert self.head is not None, "DLl Empty"
        current = min_ = self.head
        while current is not None:
            if min_.element.priority > current.element.priority:
                min_ = current
            current = current.next

        temp = min_.element.data
        if min_ is self.head:  # first element removal
            self.head = min_.next
        elif min_ is self.tail:
            self.tail = min_.prev
        else:
            min_.prev.next = min_.next
            min_.element = min_.next = min_.prev = None
        self.size -= 1
        return temp


class BoundedPriorityQueueArray:
    """ Limitation of using priority of an element
    Suppose, if PQ maxLength is 5 then you should only
    able to provide priority upto maxLength -> [0, p].
    Also, the enqueue operation is limited too
    """

    def __init__(self, maxLength):
        self.maxLength = maxLength
        self.array = [None] * maxLength
        for i in range(self.maxLength):
            self.array[i] = LinkedQueue()
        self.size = 0

    def is_empty(self):
        return self.size == 0

    def __len__(self):
        return self.size

    def enqueue(self, key, item):
        """ enqueue in the end
        time- O(1)
        """

        assert item < self.maxLength, "Priority int should be lesser"
        self.array[item].FirstIn(key)
        self.size += 1
        return self

    def dequeue(self):
        """ Search the element having highest priority
        time- O(n)
        """

        if self.is_empty():
            raise Exception(" PQ empty")
        i = 0
        while i < self.maxLength:
            if self.array[i].is_empty() is False:
                break
            i += 1

        temp = self.array[i].FirstOut()
        self.size -= 1
        return temp