-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathevent_queue.py
More file actions
63 lines (47 loc) · 1.7 KB
/
event_queue.py
File metadata and controls
63 lines (47 loc) · 1.7 KB
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
58
59
60
61
62
63
"""
A priority queue for events.
"""
from events import Event
class EventQueue:
"""A priority queue of events.
Events are dequeued in timestamp order (earlier timestamp = higher priority).
"""
def is_empty(self) -> bool:
"""Return whether this event queue contains no items."""
raise NotImplementedError
def enqueue(self, event: Event) -> None:
"""Add event to this event queue, sorted by its timestamp."""
raise NotImplementedError
def dequeue(self) -> Event:
"""Remove and return the earliest event in this event queue.
Preconditions:
- not self.is_empty()
"""
raise NotImplementedError
class EventQueueList(EventQueue):
"""A queue of events that can be dequeued in timestamp order.
Note: this is related to the "PriorityQueueSorted" class
we discussed last week.
"""
# Private Instance Attributes:
# _events: a list of the events in this queue
_events: list[Event]
def __init__(self) -> None:
"""Initialize a new and empty event queue."""
self._events = []
def is_empty(self) -> bool:
"""Return whether this event queue contains no items."""
return self._events == []
def enqueue(self, event: Event) -> None:
"""Add event to this event queue."""
index = 0
while index < len(self._events) and \
self._events[index].timestamp > event.timestamp:
index = index + 1
self._events.insert(index, event)
def dequeue(self) -> Event:
"""Remove and return the earliest event in this event queue.
Preconditions:
- not self.is_empty()
"""
return self._events.pop()