-
Notifications
You must be signed in to change notification settings - Fork 0
/
ring_buffer2.py
58 lines (49 loc) · 1.63 KB
/
ring_buffer2.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class RingBuffer:
def __init__(self, size):
self.size = size
self.buffer = [None] * size
self.head = 0
self.tail = 0
self.count = 0
def is_full(self):
return self.count == self.size
def is_empty(self):
return self.count == 0
def enqueue(self, value):
if self.is_full():
print("Buffer is full. Overwriting oldest data.")
self.tail = (self.tail + 1) % self.size
else:
self.count += 1
self.buffer[self.head] = value
self.head = (self.head + 1) % self.size
def dequeue(self):
if self.is_empty():
raise IndexError("Buffer is empty")
value = self.buffer[self.tail]
self.buffer[self.tail] = None
self.tail = (self.tail + 1) % self.size
self.count -= 1
return value
def __repr__(self):
return f"RingBuffer({self.buffer})"
# Exemple d'utilisation du RingBuffer
if __name__ == "__main__":
rb = RingBuffer(5)
while True:
action = input("Choisissez une action (enqueue, dequeue, quit) : ").strip().lower()
if action == "enqueue":
value = int(input("Entrez une valeur à ajouter : "))
rb.enqueue(value)
print(rb)
elif action == "dequeue":
try:
value = rb.dequeue()
print(f"Dequeued: {value}")
print(rb)
except IndexError as e:
print(e)
elif action == "quit":
break
else:
print("Action non valide. Veuillez choisir 'enqueue', 'dequeue' ou 'quit'.")