dsa.queue

  1""" Module containing queue classes. """
  2from dsa.array import CircularArray
  3
  4class Queue:
  5    """ 
  6    A static queue implementation. 
  7    """
  8    def __init__(self, contents=None, capacity: int=10):
  9        """ 
 10        Initialize the queue with a given capacity.
 11
 12        Args:
 13            contents: The list with contents to enqueue.
 14            capacity: The initial size of the stack (defaults to 10).
 15        """
 16        self._array = [None] * capacity
 17        self._front = 0
 18        #: number of elements in queue
 19        self.count = 0
 20
 21        if contents:
 22            for e in contents:
 23                self.enqueue(e)
 24    
 25    def enqueue(self, element):
 26        """
 27        Enqueue an element into the queue. Raise Exception when trying to enqueue more elements than the capacity.
 28
 29        Args:
 30            element: The element to enqueue.
 31
 32        Raises:
 33            Exception: When trying to enqueue more elements than the capacity.
 34
 35        Returns:
 36            None
 37        """
 38        if self.count >= len(self._array):
 39            raise Exception("Capacity Reached")
 40
 41        index = (self._front + self.count) % len(self._array)
 42        self._array[index] = element
 43        self.count += 1
 44        
 45    def dequeue(self):
 46        """
 47        Dequeue an element from the queue. Raise Exception when there are no elements to dequeue.
 48
 49        Raises:
 50            Exception: When there are no elements to dequeue.
 51
 52        Returns:
 53            The from element in the queue.
 54        """
 55        if self.is_empty():
 56            raise Exception("Empty Queue")
 57
 58        element = self._array[self._front]
 59        self._front += 1
 60        if self._front >= len(self._array):
 61            self._front = 0
 62        self.count -= 1
 63
 64        return element
 65    
 66    def peek(self):
 67        """
 68        Return the element in front of the queue. Raise Exception if queue is empty.
 69
 70        Returns:
 71            The element in front of the queue.
 72
 73        Raises:
 74            Exception: When the queue is empty.
 75        """
 76        if self.is_empty():
 77            raise Exception("Empty Queue")
 78
 79        return self._array[self._front]
 80
 81    def is_empty(self):
 82        """
 83        Return a Boolean on whether the stack is empty or not.
 84
 85        Returns:
 86            True if the stack is empty, False otherwise.
 87        """
 88        return self.count == 0
 89    
 90    def capacity(self):
 91        """
 92        Return the capacity of the queue.
 93
 94        Returns:
 95            The capacity of the queue.
 96        """
 97        return len(self._array)
 98    
 99    @classmethod
100    def from_list(cls, alist):
101        """
102        Set the contents of a queue into an array. Raise Exception when trying to enqueue more elements than the capacity.
103
104        Args:
105            alist: The list with contents to enqueue.
106
107        Returns:
108            The queue with the contents of the list.
109        """
110        q = cls()
111        for e in alist:
112            q.enqueue(e)
113        return q
114
115    def to_ordered_list(self) -> list:
116        """
117        Return the contents of the queue as a Python list.
118
119        Returns:
120            The contents of the queue as a Python list.
121        """
122        if self._front + self.count <= len(self._array):
123            return self._array[self._front:self._front + self.count]
124        else:
125            end_part = self._array[self._front:]
126            start_part = self._array[:self._front + self.count - len(self._array)]
127            return end_part + start_part
128        
129    def raw_view(self):
130        """
131        Return the queue in its array representation.
132
133        Returns:
134            The array representation of the queue.
135        """
136        return self._array
137
138    def __repr__(self):
139        """
140        Return a string with details of the queue.
141
142        Returns:
143            A string with details of the queue.
144        """
145        ordered_list = self.to_ordered_list()
146        # arr = []
147        # for i in range(self.count):
148        #     index = (i + self._front) % len(self._array)
149        #     arr.append(str(self._array[index]))
150        arrstr = ", ".join([str(e) for e in ordered_list])
151        return f"[{arrstr}] count: {self.count} Capacity: {self.capacity()}"
152    
153    def __len__(self):
154        """
155        Return the count of items in the queue.
156
157        Returns:
158            The count of items in the queue.
159        """
160        return self.count
161
162    def __eq__(self, other):
163        """
164        Compare two queues (static/dynamic/circular) for value-based equality.
165
166        Returns:
167            True if both are Queue, DynamicQueue, or CircularQueue instances and their contents are equal.
168            For non-queue types, returns NotImplemented.
169        """
170        if isinstance(other, (Queue, DynamicQueue, CircularQueue)):
171            def _as_list(obj):
172                # Queue/DynamicQueue expose to_ordered_list; CircularQueue exposes to_list
173                if hasattr(obj, 'to_ordered_list'):
174                    return obj.to_ordered_list()
175                if hasattr(obj, 'to_list'):
176                    return obj.to_list()
177                return None
178            return _as_list(self) == _as_list(other)
179        return NotImplemented
180    
181
182class DynamicQueue(Queue):
183    """ 
184    A dynamic queue implementation. Note that shrink is not impelmented.
185    """
186    def __init__(self, contents=None, capacity: int=10):
187        """
188        Initialize the queue with a given capacity.
189        
190        Args:
191            contents: The list with contents to enqueue.
192            capacity: The initial size of the stack (defaults to 10).
193        """
194        super().__init__(contents, capacity)
195    
196    def grow(self):
197        """ 
198        Double the capacity of the current array.
199
200        Returns:
201            None
202        """
203        new_array = [ None ] * len(self._array) * 2
204        
205        # copy elements
206        for i in range(self.count):
207            new_array[i] = self._array[i + self._front]
208        self._front = 0
209        self._array = new_array
210
211    def check_capacity(self):
212        """ 
213        If count >= capacity, grow the array.
214
215        Returns:
216            None
217        """
218        if self._front + self.count >= len(self._array):
219            self.grow()
220
221    def enqueue(self, element):
222        """
223        Enqueue an element into the queue. Increae capacity if count is greater than the capacity.
224
225        Args:
226            element: the element to enqueue
227
228        Returns:
229            None
230        """
231        self.check_capacity()
232        index = self._front + self.count
233        self._array[index] = element
234        self.count += 1
235
236class CircularQueue(CircularArray):
237    """ 
238    A circular queue implementation. 
239    """
240    def __init__(self, contents=None, capacity: int=10):
241        """
242        Initialize the queue with a given capacity.
243
244        Args:
245            contents: The list with contents to enqueue.
246            capacity: The initial size of the stack (defaults to 10).
247        """
248        super().__init__(None, capacity)
249        # self._start is equivalent to the queue's front index
250
251        if contents:
252            for e in contents:
253                self.enqueue(e)
254
255    def enqueue(self, element):
256        """
257        Enqueue an element into the queue. Wrap around when trying to enqueue more elements than the capacity.
258
259        Args:
260            element: The element to enqueue.
261
262        Returns:
263            None
264        """
265        return super().append(element)
266
267    def dequeue(self):
268        """
269        Dequeue an element from the queue. Raise Exception when there are no elements to dequeue.
270
271        Raises:
272            Exception: When there are no elements to dequeue.
273
274        Returns:
275            The from element in the queue.
276        """
277        if self.is_empty():
278            raise Exception("Empty Queue")
279
280        element = self._array[self._start]
281        self._start = (self._start + 1) % len(self._array)
282        self.count -= 1
283        return element
284
285    def peek(self):
286        """
287        Return the element in front of the queue. Raise Exception if queue is empty.
288
289        Returns:
290            The element in front of the queue.
291
292        Raises:
293            Exception: When the queue is empty.
294        """
295        if self.is_empty():
296            raise Exception("Empty Queue")
297
298        return self._array[self._start]