This commit is contained in:
bt3gl 2023-07-30 21:40:09 -07:00
parent 1d44d182e2
commit a85ed914d3
320 changed files with 0 additions and 0 deletions

120
queues/Queue.py Normal file
View file

@ -0,0 +1,120 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# author: bt3gl
import heapq
class Queue:
def __init__(self):
self._in = []
self._out = []
######################
# Private methods
######################
def _transfer_in_to_out(self):
while self._in:
self._out.append(self._in.pop())
def __repr__(self):
if not self._out:
self._transfer_in_to_out()
return f'{self._out}'
######################
# Properties
######################
@property
def size(self):
return len(self._in) + len(self._out)
@property
def peek(self):
if not self._out:
self._transfer_in_to_out()
if self._out:
return self._out[-1]
else:
print('❌ Queue is empty, cannot peek.')
@property
def is_empty(self):
return not (bool(self._in) or bool(self._out))
######################
# Public methods
######################
def enqueue(self, item):
self._in.append(item)
def dequeue(self):
if not self._out:
while self._in:
self.out.append(self._in.pop())
if self._out:
self._out.pop()
else:
print('❌ Queue is empty, cannot dequeue.')
class PriorityQueue:
def __init__(self):
self.queue = []
self.index = 0
def push(self, item, priority):
heapq.heappush(self.queue, (-priority, self.index, item))
self.index += 1
def pop(self):
return heapq.heappop(self.queue)[-1]
class Item:
def __init__(self, name):
self.name = name
def __repr__(self):
return f'{self.name}'
if __name__ == '__main__':
######################
# Simple Queue
######################
print('🧪 Testing Queue...')
queue = Queue()
print(f"Is the queue empty? {queue.is_empty}")
print("Adding 1 to 10 in the queue...")
for i in range(1, 11):
queue.enqueue(i)
print(f"Queue: {queue}")
print(f"\nQueue size: {queue.size}")
print(f"Queue peek : {queue.peek}")
print(f"Is the queue empty? {queue.is_empty}")
print(f"\nDequeue...")
queue.dequeue()
print(f"Queue: {queue}")
print(f"\nQueue size: {queue.size}")
print(f"Queue peek: {queue.peek}")
print(f"Is the queue empty? {queue.is_empty}")
######################
# Priority Queue
######################
print('\n\n🧪 Testing Priority Queue...')
q = PriorityQueue()
q.push(Item('Item 1'), 1)
q.push(Item('Item 4'), 4)
q.push(Item('Item 3'), 3)
print(f"Priority Queue: {q.queue}")
print(f"Pop: {q.pop()}")
print(f"Priority Queue: {q.queue}")

119
queues/circular_queue.py Normal file
View file

@ -0,0 +1,119 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# author: bt3gl
## implement a circular queue
class CircularQueue:
def __init__(self, k: int):
self.head = 0
self.tail = 0
self.size = k
self.queue = [None] * self.size
def enQueue(self, value: int) -> bool:
if value is None:
return False
if self.isFull():
return False
if self.isEmpty():
self.heard = 0
while self.queue[self.tail] != None:
self.tail += 1
if self.tail == self.size:
self.tail = 0
self.queue[self.tail] = value
return True
def deQueue(self) -> bool:
if self.isEmpty():
return False
value = self.queue[self.head]
self.queue[self.head] = None
self.head += 1
if self.head == self.size:
self.head = 0
return True
def Front(self) -> int:
return self.queue[self.head] or -1
def Rear(self) -> int:
return self.queue[self.tail] or -1
def isEmpty(self) -> bool:
for n in self.queue:
if n is not None:
return False
return True
def isFull(self) -> bool:
for n in self.queue:
if n is None:
return False
return True
if __name__ == "__main__":
q = CircularQueue(5)
print(f'q: {q.queue}')
print(f'q.isEmpty(): {q.isEmpty()}')
print(f'q.isFull(): {q.isFull()}')
print(f'q.enQueue(1): {q.enQueue(1)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.enQueue(2): {q.enQueue(2)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.enQueue(3): {q.enQueue(3)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.enQueue(4): {q.enQueue(4)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'isFull(): {q.isFull()}')
print(f'q.enQueue(5): {q.enQueue(5)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'isFull(): {q.isFull()}')
print(f'q.isEmpty(): {q.isEmpty()}')
print(f'q: {q.queue}')
print(f'q.enQueue(6): {q.enQueue(6)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'isFull(): {q.isFull()}')
print(f'q.isEmpty(): {q.isEmpty()}')
print()
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.Front(): {q.Front()}')
print(f'q.isEmpty(): {q.isEmpty()}')
print(f'q.isFull(): {q.isFull()}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.Front(): {q.Front()}')
print(f'q.isEmpty(): {q.isEmpty()}')
print(f'q.isFull(): {q.isFull()}')

111
queues/circular_queue_II.py Normal file
View file

@ -0,0 +1,111 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# author: bt3gl
## implement a circular queue
class CircularQueue:
def __init__(self, k: int):
self.head = -1
self.tail = -1
self.size = k
self.queue = [None] * self.size
def enQueue(self, value: int) -> bool:
if self.isFull():
return False
if self.isEmpty():
head = 0;
self.tail = (self.tail + 1) % self.size
self.queue[self.tail] = value
return True
def deQueue(self) -> bool:
if self.isEmpty():
return False
if self.head == self.tail:
self.head = -1
self.tail = -1
return True
self.head = (self.head + 1) % self.size
return True
def Front(self) -> int:
if self.isEmpty():
return -1
return self.queue[self.head]
def Rear(self) -> int:
if self.isEmpty():
return -1
return self.queue[self.tail]
def isEmpty(self) -> bool:
return self.head == -1
def isFull(self) -> bool:
return (self.tail + 1) % self.size == self.head
if __name__ == "__main__":
q = CircularQueue(5)
print(f'q: {q.queue}')
print(f'q.isEmpty(): {q.isEmpty()}')
print(f'q.isFull(): {q.isFull()}')
print(f'q.enQueue(1): {q.enQueue(1)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.enQueue(2): {q.enQueue(2)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.enQueue(3): {q.enQueue(3)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.enQueue(4): {q.enQueue(4)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'isFull(): {q.isFull()}')
print(f'q.enQueue(5): {q.enQueue(5)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'isFull(): {q.isFull()}')
print(f'q.isEmpty(): {q.isEmpty()}')
print(f'q: {q.queue}')
print(f'q.enQueue(6): {q.enQueue(6)}')
print(f'h: {q.head}, t: {q.tail}')
print(f'isFull(): {q.isFull()}')
print(f'q.isEmpty(): {q.isEmpty()}')
print()
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.Front(): {q.Front()}')
print(f'q.isEmpty(): {q.isEmpty()}')
print(f'q.isFull(): {q.isFull()}')
print(f'q.deQueue(): {q.deQueue()}')
print(f'q: {q.queue}')
print(f'h: {q.head}, t: {q.tail}')
print(f'q.Front(): {q.Front()}')
print(f'q.isEmpty(): {q.isEmpty()}')
print(f'q.isFull(): {q.isFull()}')

View file

@ -0,0 +1,28 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# author: bt3gl
class Logger:
def __init__(self):
self.msg_set = set()
self.msg_queue = deque()
def print_message(self, timestamp: int, message: str) -> bool:
while self.msg_queue:
msg, msg_timestamp = self.msg_queue[0]
if timestamp - msg_timestamp >= 10:
self.msg_queue.popleft()
self.msg_set.remove(msg)
else:
break
if message not in self.msg_set:
self.msg_set.add(message)
self.msg_queue.append((message, timestamp))
return True
else:
return False

29
queues/moving_average.py Normal file
View file

@ -0,0 +1,29 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# author: bt3gl
'''
Given a stream of integers and a window size, calculate the moving average of all integers in the sliding window.
Implement the MovingAverage class:
MovingAverage(int size) Initializes the object with the size of the window size.
double next(int val) Returns the moving average of the last size values of the stream.
'''
class MovingAverage:
def __init__(self, size: int):
self.queue = []
self.size = size
def next(self, val: int) -> float:
self.queue.append(val)
if len(self.queue) > self.size:
self.queue.pop(0)
return sum(self.queue) / len(self.queue)