Queues
Download exercises zip
Introduction
In these exercises, you will be implementing several queues.
What to do
unzip exercises in a folder, you should get something like this:
queues
queues.ipynb
circular_queue.py
circular_queue_test.py
circular_queue_sol.py
jupman.py
sciprog.py
open the editor of your choice (for example Visual Studio Code, Spyder or PyCharme), you will edit the files ending in
.pyfilesGo on reading this notebook, and follow instuctions inside.
1. LinkedQueue
Open linked_queue.py.
You are given a queue implemented as a linked list, with usual _head pointer plus additional _tail pointer and _size counter
Data in enqueued at the right, in the tail
Data is dequeued at the left, removing it from the head
Example, where the arrows represent _next pointers:
_head _tail
a -> b -> c -> d -> e -> f
In this exercise you will implement the methods enqn(lst) and deqn(n) which respectively enqueue a python list of n elements and dequeue n elements, returning python a list of them.
Here we show an example usage, see to next points for detailed instructions.
Example:
[1]:
from linked_queue_sol import *
[2]:
q = LinkedQueue()
[3]:
print(q)
LinkedQueue:
[4]:
q.enqn(['a','b','c'])
Returns nothing, queue becomes:
_head _tail
a -> b -> c
[5]:
q.enqn(['d'])
Returns nothing, queue becomes:
_head _tail
a -> b -> c -> d
[6]:
q.enqn(['e','f'])
Returns nothing, queue becomes:
_head _tail
a -> b -> c -> d -> e -> f
[7]:
q.deqn(3)
[7]:
['a', 'b', 'c']
Returns [‘d’, ‘e’, ‘f’] and queue becomes:
_head _tail
a -> b -> c
[8]:
q.deqn(1)
[8]:
['d']
Returns ['c'] and queue becomes:
_head _tail
a -> b
q.deqn(5)
---------------------------------------------------------------------------
LookupError Traceback (most recent call last)
<ipython-input-55-e68c2e9949d0> in <module>()
1
----> 2 q.deqn(5)
~/Da/prj/sciprog-ds/prj/queues/linked_queue_sol.py in deqn(self, n)
202 #jupman-raise
203 if n > self._size:
--> 204 raise LookupError('Asked to dequeue %s elements, but only %s are available!' % (n, self._size))
205
206 ret = []
LookupError: Asked to dequeue 5 elements, but only 2 are available!
Raises LookupError as there aren’t enough elements to remove
1.1 enqn
Implement the method enqn:
def enqn(self, lst):
""" Enqueues provided list of elements at the tail of the queue
- Required complexity: O(len(lst))
- NOTE: remember to update the _size and _tail
Example: supposing arrows represent _next pointers:
_head _tail
a -> b -> c
Calling
q.enqn(['d', 'e', 'f', 'g'])
will produce the queue:
_head _tail
a -> b -> c -> d -> e -> f -> g
Testing: python3 -m unittest linked_queue_test.EnqnTest
1.2 deqn
Implement the method deqn:
def deqn(self, n):
""" Removes n elements from the head, and return them as a Python list,
where the first element that was enqueued will appear at the
*beginning* of the returned Python list.
- if n is greater than the size of the queue, raises a LookupError.
- required complexity: O(n)
NOTE 1: return a list of the *DATA* in the nodes, *NOT* the nodes
themselves
NOTE 2: DO NOT try to convert the whole queue to a Python
list for playing with splices.
NOTE 3: remember to update _size, _head and _tail when needed.
For example, supposing arrows represent _next pointers:
_head _tail
a -> b -> c -> d -> e -> f -> g
q.deqn(3) will return the Python list ['a', 'b', 'c']
After the call, the queue will be like this:
_head _tail
d -> e -> f -> g
"""
Testing: python3 -m unittest linked_queue_test.DeqnTest
2. CircularQueue
We moved this section to a separate worksheet
3. ItalianQueue
You will implement an ItalianQueue, modelled as a LinkedList with two pointers, a _head and a _tail.
an element is enqueued scanning from
_headuntil a matching group is found, in which case are inserted after (that is, at the right) of the matching group, otherwise the element is appended at the_tailan element is dequeued from the
_head
3.1 Slow v1
To gain some understanding about the data structure, look at the following excerpts.
Excerpt from Node:
class Node:
""" A Node of an ItalianQueue.
Holds both data and group provided by the user.
"""
def __init__(self, initdata, initgroup):
def get_data(self):
def get_group(self):
def get_next(self):
# etc ..
Excerpt from ItalianQueue class:
class ItalianQueue:
""" An Italian queue, v1.
- Implemented as a LinkedList
- Worst case enqueue is O(n)
- has extra methods, for accessing groups and tail:
- top_group()
- tail()
- tail_group()
Each element is assigned a group; during enqueing, queue is scanned
from head to tail to find if there is another element with a
matching group.
- If there is, element to be enqueued is inserted after the last
element in the same group sequence (that is, to the right of
the group)
- otherwise the element is inserted at the end of the queue
"""
def __init__(self):
""" Initializes the queue. Note there is no capacity as parameter
- MUST run in O(1)
"""
Example:
[9]:
from italian_queue_sol import *
q = ItalianQueue()
print(q)
ItalianQueue:
_head: None
_tail: None
[10]:
q.enqueue('a','x') # 'a' is the element,'x' is the group
[11]:
print(q)
ItalianQueue: a
x
_head: Node(a,x)
_tail: Node(a,x)
[12]:
q.enqueue('c','y') # 'c' belongs to new group 'y', goes to the end of the queue
[13]:
print(q)
ItalianQueue: a->c
x y
_head: Node(a,x)
_tail: Node(c,y)
[14]:
q.enqueue('d','y') # 'd' belongs to existing group 'y', goes to the end of the group
[15]:
print(q)
ItalianQueue: a->c->d
x y y
_head: Node(a,x)
_tail: Node(d,y)
[16]:
q.enqueue('b','x') # 'b' belongs to existing group 'x', goes to the end of the group
[17]:
print(q)
ItalianQueue: a->b->c->d
x x y y
_head: Node(a,x)
_tail: Node(d,y)
[18]:
q.enqueue('f','z') # 'f' belongs to new group, goes to the end of the queue
[19]:
print(q)
ItalianQueue: a->b->c->d->f
x x y y z
_head: Node(a,x)
_tail: Node(f,z)
[20]:
q.enqueue('e','y') # 'e' belongs to an existing group 'y', goes to the end of the group
[21]:
print(q)
ItalianQueue: a->b->c->d->e->f
x x y y y z
_head: Node(a,x)
_tail: Node(f,z)
[22]:
q.enqueue('g','z') # 'g' belongs to an existing group 'z', goes to the end of the group
[23]:
print(q)
ItalianQueue: a->b->c->d->e->f->g
x x y y y z z
_head: Node(a,x)
_tail: Node(g,z)
[24]:
q.enqueue('h','z') # 'h' belongs to an existing group 'z', goes to the end of the group
[25]:
print(q)
ItalianQueue: a->b->c->d->e->f->g->h
x x y y y z z z
_head: Node(a,x)
_tail: Node(h,z)
Dequeue is always from the head, without taking in consideration the group:
[26]:
q.dequeue()
[26]:
'a'
[27]:
print(q)
ItalianQueue: b->c->d->e->f->g->h
x y y y z z z
_head: Node(b,x)
_tail: Node(h,z)
[28]:
q.dequeue()
[28]:
'b'
[29]:
print(q)
ItalianQueue: c->d->e->f->g->h
y y y z z z
_head: Node(c,y)
_tail: Node(h,z)
[30]:
q.dequeue()
[30]:
'c'
[31]:
print(q)
ItalianQueue: d->e->f->g->h
y y z z z
_head: Node(d,y)
_tail: Node(h,z)
3.1.1 init
Implement methods in file italian_queue.py in the order they are presented up until enqueue excluded
Testing: python3 -m unittest italian_queue_test.InitEmptyTest
3.1.2 Slow enqueue
Implement version 1 of enqueue running in \(O(n)\) where \(n\) is the queue size.
def enqueue(self, v, g):
""" Enqueues provided element v having group g, with the following
criteria:
Queue is scanned from head to find if there is another element
with a matching group:
- if there is, v is inserted after the last element in the
same group sequence (so to the right of the group)
- otherwise v is inserted at the end of the queue
- MUST run in O(n)
"""
Testing: python3 -m unittest italian_queue_test.EnqueueTest
QUESTION: The ItalianQueue was implemented as a LinkedList. Even if this time we don’t care much about perfomance, if we wanted an efficient enqueue operation, could we start with a circular data structure ? Or would you prefer improving a LinkedList ?
3.1.2 dequeue
Implement version 1 of dequeue running in \(O(1)\)
def dequeue(self):
""" Removes head element and returns it.
- If the queue is empty, raises a LookupError.
- MUST run in O(1)
"""
Testing: python3 -m unittest italian_queue_test.DequeueTest
3.2 Fast v2
3.2.1 Save a copy
You already wrote a lot of code, and you don’t want to lose it, right? Since we are going to make many modifications, when you reach a point when the code does something useful, it is good practice to save a copy of what you have done somewhere, so if you later screw up something, you can always restore the copy.
Copy the whole folder
queuesin a new folderqueues_v1Add also in the copied folder a separate
README.txtfile, writing inside the version (like1.0), the date, and a description of the main features you implemented (for example “Simple Italian Queue, not particularly performant”).Backing up the work is a form of the so-called versioning : there are much better ways to do it (like using git) but we don’t address them here.
WARNING: DO NOT SKIP THIS STEP!
No matter how smart you are, you will fail, and a backup may be the only way out.
WARNING: NOT CONVINCED YET?
If you still don’t understand why you should spend time with this copy bureaucracy, to help you enter the right mood imagine tomorrow is demo day with your best client and you screw up the only working version: your boss will skin you alive.
3.2.2 make it fast
Improve enqueue adn dequeue so they work both in \(O(1)\)
HINT:
You will need an extra data structure that keeps track of the starting points of each group and how they are ordered
You will also need to update this data structure as
enqueueanddequeuecalls are made
SOLUTION: This exercise was proposed during this exam
4. Supermarket queues
In this exercises, you will try to model a supermarket containing several cash queues.
CashQueue
WARNING: DO *NOT* MODIFY CashQueue CLASS
For us, a CashQueue is a simple queue of clients represented as strings. A CashQueue supports the enqueue, dequeue, size and is_empty operations:
Clients are enqueued at the right, in the tail
Clients are dequeued from the left, removing them from the head
For example:
q = CashQueue()
q.is_empty() # True
q.enqueue('a') # a
q.enqueue('b') # a,b
q.enqueue('c') # a,b,c
q.size() # 3
q.dequeue() # returns: a
# queue becomes: [b,c]
q.dequeue() # returns: b
# queue becomes: [c]
q.dequeue() # returns: c
# queue becomes: []
q.dequeue() # raises LookupError as there aren't enough elements to remove
Supermarket
A Supermarket contains several cash queues. It is possible to initialize a Supermarket by providing queues as simple python lists, where the first clients arrived are on the left, and the last clients are on the right.
For example, by calling:
s = Supermarket([
['a','b','c'], # <------ clients arrive from right
['d'],
['f','g']
])
internally three CashQueue objects are created. Looking at the first queue with clients ['a','b','c'], a at the head arrived first and c at the tail arrived last
>>> print(s)
Supermarket
0 CashQueue: ['a', 'b', 'c']
1 CashQueue: ['d']
2 CashQueue: ['f', 'g']
Note a supermarket must have at least one queue, which may be empty:
s = Supermarket( [[]] )
>>> print(s)
Supermarket
0 CashQueue: []
Supermarket as a queue
Our Supermarket should maximize the number of served clients (we assume each clients is served in an equal amount of time). To do so, the whole supermarket itself can be seen as a particular kind of queue, which allows the enqueue and dequeue operations described as follows:
by calling
supermarket.enqueue(client)a client gets enqueued in the shortestCashQueue.by calling
supermarket.dequeue(), all clients which are at the heads of non-emptyCashQueues are dequeued all at once, and their list is returned (this simulates parallelism).
Implementation
Now start editing supermarket.py implementing methods in the following points.
4.1 Supermarket size
Implement Supermarket.size :
def size(self):
""" Return the total number of clients present in all cash queues.
"""
Testing: python3 -m unittest supermarket_test.SizeTest
4.2 Supermarket dequeue
Implement Supermarket.dequeue :
def dequeue(self):
""" Dequeue all the clients which are at the heads of non-empty cash queues,
and return a list of such clients.
- clients are returned in the same order as found in the queues
- if supermarket is empty, an empty list is returned
For example, suppose we have following supermarket:
0 ['a','b','c']
1 []
2 ['d','e']
3 ['f']
A call to deque() will return ['a','d','f']
and the supermarket will now look like this:
0 ['b','c']
1 []
2 ['e']
3 []
"""
Testing: python3 -m unittest supermarket_test.DequeueTest
4.3 Supermarket enqueue
Implement Supermarket.enqueue :
def enqueue(self, client):
""" Enqueue provided client in the cash queue with minimal length.
If more than one minimal length cash queue is available, the one
with smallest index is chosen.
For example:
If we have supermarket
0 ['a','b','c']
1 ['d','e','f','g']
2 ['h','i']
3 ['m','n']
since queues 2 and 3 have both minimal length 2,
supermarket.enqueue('z') will enqueue the client on queue 2:
0 ['a','b','c']
1 ['d','e','f','g']
2 ['h','i','z']
3 ['m','n']
"""
Testing: python3 -m unittest supermarket_test.EnqueueTest
5. Shopping mall queues
In this exercises, you will try to model a shopping mall containing several shops and clients.
Client
WARNING: DO *NOT* MODIFY Client CLASS
For us, a Client is composed by a name (in the exercise we will use a, b, c …) and a list of shops he wants to visit as a list. We will identify the shops with letters such as x, y, z …
Note: shops to visit are a Python list intended as a stack, so the first shop to visit is at end (top) of the list
Example:
c = Client('f', ['y','x','z'])
creates a Client named f who wants to visit first the shop z, then x and finally y
Methods:
>>> print(c.name())
a
>>> print(c.to_visit())
['z','x','y']
Shop
WARNING: DO *NOT* MODIFY Shop CLASS
For us, a Shop is a class with a name and a queue of clients. A Shop supports the name, enqueue, dequeue, size and is_empty operations:
Clients are enqueued at the right, in the tail
Clients are dequeued from the left, removing them from the head
For example:
s = Shop('x') # creates a shop named 'x'
print(s.name()) # prints x
s.is_empty() # True
s.enqueue('a') # a enqueues client 'a'
s.enqueue('b') # a,b
s.enqueue('c') # a,b,c
s.size() # 3
s.dequeue() # returns: a
# queue becomes: [b,c]
s.dequeue() # returns: b
# queue becomes: [c]
s.dequeue() # returns: c
# queue becomes: []
s.dequeue() # raises LookupError as there aren't enough elements to remove
Mall
A shopping Mall contains several shops and clients. It is possible to initialize a Mall by providing
shops as a list of values
shop name , client list, where the first clients arrived are on the left, and the last clients are on the right.clients as a list of values
client name , shop to visit list
For example, by calling:
m = Mall(
[
'x', ['a','b','c'], # <------ clients arrive from right
'y', ['d'],
'z', ['f','g']
],
[
'a',['y','x'],
'b',['x'],
'c',['x'],
'd',['z','y'], # IMPORTANT: shops to visit stack grows from right, so
'f',['y','x','z'], # client 'f' wants to visit first shop 'z', then 'x', and finally 'y'
'g',['x','z']
])
Internally:
three
Shopobjects are created in anOrderedDict. Looking at the first queue with clients['a','b','c'],aat the head arrived first andcat the tail arrived last.6
Clientobjects are created in anOrderedDict. Note if a client is in a particular shop queue, that shop must be his top desired shop to visit in its stack.
>>> print(s)
Mall
Shop x: ['a', 'b', 'c']
Shop y: ['d']
Shop z: ['f', 'g']
Client a: ['y','x']
Client b: ['x']
Client c: ['x']
Client d: ['z','y']
Client f: ['x','y','z']
Client g: ['x','z']
Methods:
>>> m.shops()
OrderedDict([
('x', Shop x: ['a', 'b', 'c'])
('y', Shop y: ['d'])
('z', Shop z: ['f', 'g'])
])
>>> m.clients()
OrderedDict([
('a', Client a: ['y','x']),
('b', Client b: ['x']),
('c', Client c: ['x']),
('d', Client d: ['z','y']),
('f', Client f: ['x','y','z']),
('g', Client g: ['x','z'])
])
Note a mall must have at least one shop and may have zero clients:
m = Mall( {'x':[]}, {} )
>>> print(m)
Mall
Shop x: []
Mall as a queue
Our Mall should maximize the number of served clients (we assume each clients is served in an equal amount of time). To do so, the whole mall itself can be seen as a particular kind of queue, which allows the enqueue and dequeue operations described as follows:
by calling
mall.enqueue(client)a client gets enqueued in the topShophe wants to visit (its desired shop to visit list doesn’t change)by calling
mall.dequeue()all clients which are at the heads of non-empty
Shops are dequeued all at oncetheir top desired shop to visit is removed
if a client has any shop to visit left, he is automatically enqueued in that
Shopthe list of clients with no shops to visit is returned (this simulates parallelism)
Implementation
Now start editing mall.py implementing methods in the following points.
6.1 Mall enqueue
Implement Mall.enqueue method:
def enqueue(self, client):
""" Enqueue provided client in the top shop he wants to visit
- If client is already in the mall, raise ValueError
- if client has no shop to visit, raise ValueError
- If any of the shops to visit are not in the mall, raise ValueError
For example:
If we have this mall:
Mall
Shop x: ['a','b']
Shop y: ['c']
Client a: ['y','x']
Client b: ['x']
Client c: ['x','y']
mall.enqueue(Client('d',['x','y'])) will enqueue the client in Shop y :
Mall
Shop x: ['a','b']
Shop y: ['c','d']
Client a: ['y','x']
Client b: ['x']
Client c: ['x','y']
Client d: ['x','y']
"""
Testing: python3 -m unittest mall_test.EnqueueTest
6.2 Mall dequeue
Implement Mall.dequeue method:
def dequeue(self):
""" Dequeue all the clients which are at the heads of non-empty
shop queues,enqueues clients in their next shop to visit and return
a list of names of clients that exit the mall.
In detail:
- shop list is scanned, and all clients which are at the heads
of non-empty Shops are dequeued
VERY IMPORTANT HINT: FIRST put all this clients in a list,
THEN using that list do all of the following
- for each dequeued client, his top desired shop is removed from
his visit list
- if a client has a shop to visit left, he is automatically
enqueued in that Shop
- clients are enqueued in the same order they were dequeued
from shops
- the list of clients with no shops to visit anymore
is returned (this simulates parallelism)
- clients are returned in the same order they were dequeued
from shops
- if mall has no clients, an empty list is returned
"""
Testing: python3 -m unittest mall_test.DequeueTest
For example, suppose we have following mall:
[32]:
from mall_sol import *
[33]:
m = Mall([
'x', ['a', 'b', 'c'],
'y', ['d'],
'z', ['f', 'g']
],
[
'a', ['y', 'x'],
'b', ['x'],
'c', ['x'],
'd', ['z','y'],
'f', ['y','x','z'],
'g', ['x','z']
])
[34]:
print(m)
Mall
Shop x : ['a', 'b', 'c']
Shop y : ['d']
Shop z : ['f', 'g']
Client a : ['y', 'x']
Client b : ['x']
Client c : ['x']
Client d : ['z', 'y']
Client f : ['y', 'x', 'z']
Client g : ['x', 'z']
[35]:
m.dequeue() # first call
[35]:
[]
Clients ‘a’, ‘d’ and ‘f’ change shop, the others stay in their current shop. The mall will now look like this:
[36]:
print(m)
Mall
Shop x : ['b', 'c', 'f']
Shop y : ['a']
Shop z : ['g', 'd']
Client a : ['y']
Client b : ['x']
Client c : ['x']
Client d : ['z']
Client f : ['y', 'x']
Client g : ['x', 'z']
[37]:
m.dequeue() # second call
[37]:
['b', 'a']
because client ‘b’ was top shop in the list, ‘a’ in the second, and both clients had nothing else to visit. Client ‘g’ changes shop, the others remain in their current shop.
The mall will now look like this:
[38]:
print(m) # Clients a and b are gone
Mall
Shop x : ['c', 'f', 'g']
Shop y : []
Shop z : ['d']
Client c : ['x']
Client d : ['z']
Client f : ['y', 'x']
Client g : ['x']
[39]:
m.dequeue() # third call
[39]:
['c', 'd']
[40]:
print(m)
Mall
Shop x : ['f', 'g']
Shop y : []
Shop z : []
Client f : ['y', 'x']
Client g : ['x']
[41]:
m.dequeue() # fourth call
[41]:
[]
[42]:
print(m)
Mall
Shop x : ['g']
Shop y : ['f']
Shop z : []
Client f : ['y']
Client g : ['x']
[43]:
m.dequeue() # fifth call
[43]:
['g', 'f']
[44]:
print(m)
Mall
Shop x : []
Shop y : []
Shop z : []
[45]:
m.dequeue() # no clients left
[45]:
[]
6. Company queues
We can model a company as a list of many employees ordered by their rank, the highest ranking being the first in the list. We assume all employees have different rank. Each employee has a name, a rank, and a queue of tasks to perform (as a Python deque).
When a new employee arrives, it is inserted in the list in the right position according to his rank:
[46]:
from company_sol import *
c = Company()
print(c)
Company:
name rank tasks
[47]:
c.add_employee('x',9)
[48]:
print(c)
Company:
name rank tasks
x 9 deque([])
[49]:
c.add_employee('z',2)
[50]:
print(c)
Company:
name rank tasks
x 9 deque([])
z 2 deque([])
[51]:
c.add_employee('y',6)
[52]:
print(c)
Company:
name rank tasks
x 9 deque([])
y 6 deque([])
z 2 deque([])
7.1 add_employee
Implement this method:
def add_employee(self, name, rank):
"""
Adds employee with name and rank to the company, maintaining
the _employees list sorted by rank (higher rank comes first)
Represent the employee as a dictionary with keys 'name', 'rank'
and 'tasks' (a Python deque)
- here we don't mind about complexity, feel free to use a
linear scan and .insert
- If an employee of the same rank already exists, raise ValueError
- if an employee of the same name already exists, raise ValueError
"""
Testing: python3 -m unittest company_test.AddEmployeeTest
7.2 add_task
Each employee has a queue of tasks to perform. Tasks enter from the right and leave from the left. Each task has associated a required rank to perform it, but when it is assigned to an employee the required rank may exceed the employee rank or be far below the employee rank. Still, when the company receives the task, it is scheduled in the given employee queue, ignoring the task rank.
[53]:
c.add_task('a',3,'x')
[54]:
c
[54]:
Company:
name rank tasks
x 9 deque([('a', 3)])
y 6 deque([])
z 2 deque([])
[55]:
c.add_task('b',5,'x')
[56]:
c
[56]:
Company:
name rank tasks
x 9 deque([('a', 3), ('b', 5)])
y 6 deque([])
z 2 deque([])
[57]:
c.add_task('c',12,'x')
c.add_task('d',1,'x')
c.add_task('e',8,'y')
c.add_task('f',2,'y')
c.add_task('g',8,'y')
c.add_task('h',10,'z')
[58]:
c
[58]:
Company:
name rank tasks
x 9 deque([('a', 3), ('b', 5), ('c', 12), ('d', 1)])
y 6 deque([('e', 8), ('f', 2), ('g', 8)])
z 2 deque([('h', 10)])
Implement this function:
def add_task(self, task_name, task_rank, employee_name):
""" Append the task as a (name, rank) tuple to the tasks of
given employee
- If employee does not exist, raise ValueError
"""
Testing: python3 -m unittest company_test.AddTaskTest
7.3 work
Work in the company is produced in work steps. Each work step produces a list of all task names executed by the company in that work step.
A work step is done this way:
For each employee, starting from the highest ranking one, dequeue its current task (from the left), and than compare the task required rank with the employee rank according to these rules:
When an employee discovers a task requires a rank strictly greater than his rank, he will append the task to his supervisor tasks. Note the highest ranking employee may be forced to do tasks that are greater than his rank.
When an employee discovers he should do a task requiring a rank strictly less than his, he will try to see if the next lower ranking employee can do the task, and if so append the task to that employee tasks.
When an employee cannot pass the task to the supervisor nor the next lower ranking employee, he will actually execute the task, adding it to the work step list
Example:
[59]:
c
[59]:
Company:
name rank tasks
x 9 deque([('a', 3), ('b', 5), ('c', 12), ('d', 1)])
y 6 deque([('e', 8), ('f', 2), ('g', 8)])
z 2 deque([('h', 10)])
[60]:
c.work()
DEBUG: Employee x gives task ('a', 3) to employee y
DEBUG: Employee y gives task ('e', 8) to employee x
DEBUG: Employee z gives task ('h', 10) to employee y
DEBUG: Total performed work this step: []
[60]:
[]
[61]:
c
[61]:
Company:
name rank tasks
x 9 deque([('b', 5), ('c', 12), ('d', 1), ('e', 8)])
y 6 deque([('f', 2), ('g', 8), ('a', 3), ('h', 10)])
z 2 deque([])
[62]:
c.work()
DEBUG: Employee x gives task ('b', 5) to employee y
DEBUG: Employee y gives task ('f', 2) to employee z
DEBUG: Employee z executes task ('f', 2)
DEBUG: Total performed work this step: ['f']
[62]:
['f']
[63]:
c
[63]:
Company:
name rank tasks
x 9 deque([('c', 12), ('d', 1), ('e', 8)])
y 6 deque([('g', 8), ('a', 3), ('h', 10), ('b', 5)])
z 2 deque([])
[64]:
c.work()
DEBUG: Employee x executes task ('c', 12)
DEBUG: Employee y gives task ('g', 8) to employee x
DEBUG: Total performed work this step: ['c']
[64]:
['c']
[65]:
c
[65]:
Company:
name rank tasks
x 9 deque([('d', 1), ('e', 8), ('g', 8)])
y 6 deque([('a', 3), ('h', 10), ('b', 5)])
z 2 deque([])
[66]:
c.work()
DEBUG: Employee x gives task ('d', 1) to employee y
DEBUG: Employee y executes task ('a', 3)
DEBUG: Total performed work this step: ['a']
[66]:
['a']
[67]:
c
[67]:
Company:
name rank tasks
x 9 deque([('e', 8), ('g', 8)])
y 6 deque([('h', 10), ('b', 5), ('d', 1)])
z 2 deque([])
[68]:
c.work()
DEBUG: Employee x executes task ('e', 8)
DEBUG: Employee y gives task ('h', 10) to employee x
DEBUG: Total performed work this step: ['e']
[68]:
['e']
[69]:
c
[69]:
Company:
name rank tasks
x 9 deque([('g', 8), ('h', 10)])
y 6 deque([('b', 5), ('d', 1)])
z 2 deque([])
[70]:
c.work()
DEBUG: Employee x executes task ('g', 8)
DEBUG: Employee y executes task ('b', 5)
DEBUG: Total performed work this step: ['g', 'b']
[70]:
['g', 'b']
[71]:
c
[71]:
Company:
name rank tasks
x 9 deque([('h', 10)])
y 6 deque([('d', 1)])
z 2 deque([])
[72]:
c.work()
DEBUG: Employee x executes task ('h', 10)
DEBUG: Employee y gives task ('d', 1) to employee z
DEBUG: Employee z executes task ('d', 1)
DEBUG: Total performed work this step: ['h', 'd']
[72]:
['h', 'd']
[73]:
c
[73]:
Company:
name rank tasks
x 9 deque([])
y 6 deque([])
z 2 deque([])
Now implement this method:
def work(self):
""" Performs a work step and RETURN a list of performed task names.
For each employee, dequeue its current task from the left and:
- if the task rank is greater than the rank of the
current employee, append the task to his supervisor queue
(the highest ranking employee must execute the task)
- if the task rank is lower or equal to the rank of the
next lower ranking employee, append the task to that employee
queue
- otherwise, add the task name to the list of
performed tasks to return
"""
Testing: python3 -m unittest company_test.WorkTest
7. Concert
Start editing file concert.py.
During events with lots of potential visitors such as concerts, for speeding up check-in there are at least two queues: cash queue where tickets are sold, and entrance queue for actually entering the event.
Each visitor may or may not have a ticket. Also, since people usually attend in groups (couples, families, and so on), in the queue lines each group tends to move as a whole.
In Python, we will model a Person as a class you can create like this:
[74]:
from concert_sol import *
[75]:
Person('a', 'x', False)
[75]:
Person(a,x,False)
'a' is the name, 'x' is the group, and False indicates the person doesn’t have ticket
To model the two queues, in Concert class we have these fields and methods:
class Concert:
def __init__(self):
self._cash = deque()
self._entrance = deque()
def enqc(self, person):
""" Enqueues at the cash from the right """
self._cash.append(person)
def enqe(self, person):
""" Enqueues at the entrance from the right """
self._entrance.append(person)
7.1 dequeue
✪✪✪ Implement dequeue. If you want, you can add debug prints by calling the debug function.
def dequeue(self):
""" RETURN the names of people admitted to concert
Dequeuing for the whole queue system is done in groups, that is,
with a _single_ call to dequeue, these steps happen, in order:
1. entrance queue: all people belonging to the same group at
the front of entrance queue who have the ticket exit the queue
and are admitted to concert. People in the group without the
ticket are sent to cash.
2. cash queue: all people belonging to the same group at the front
of cash queue are given a ticket, and are queued at the entrance queue
"""
Testing: python3 -m unittest concert_test.DequeueTest
Example:
[76]:
con = Concert()
con.enqc(Person('a','x',False)) # a,b,c belong to same group x
con.enqc(Person('b','x',False))
con.enqc(Person('c','x',False))
con.enqc(Person('d','y',False)) # d belongs to group y
con.enqc(Person('e','z',False)) # e,f belongs to group z
con.enqc(Person('f','z',False))
con.enqc(Person('g','w',False)) # g belongs to group w
[77]:
con
[77]:
Concert:
cash: deque([Person(a,x,False),
Person(b,x,False),
Person(c,x,False),
Person(d,y,False),
Person(e,z,False),
Person(f,z,False),
Person(g,w,False)])
entrance: deque([])
First time we dequeue, entrance queue is empty so no one enters concert, while at the cash queue people in group x are given a ticket and enqueued at the entrance queue
NOTE: The messages on the console are just debug print, the function dequeue only return name sof people admitted to concert
[78]:
con.dequeue()
DEBUG: DEQUEUING ..
DEBUG: giving ticket to a (group x)
DEBUG: giving ticket to b (group x)
DEBUG: giving ticket to c (group x)
DEBUG: Concert:
cash: deque([Person(d,y,False),
Person(e,z,False),
Person(f,z,False),
Person(g,w,False)])
entrance: deque([Person(a,x,True),
Person(b,x,True),
Person(c,x,True)])
[78]:
[]
[79]:
con.dequeue()
DEBUG: DEQUEUING ..
DEBUG: a (group x) admitted to concert
DEBUG: b (group x) admitted to concert
DEBUG: c (group x) admitted to concert
DEBUG: giving ticket to d (group y)
DEBUG: Concert:
cash: deque([Person(e,z,False),
Person(f,z,False),
Person(g,w,False)])
entrance: deque([Person(d,y,True)])
[79]:
['a', 'b', 'c']
[80]:
con.dequeue()
DEBUG: DEQUEUING ..
DEBUG: d (group y) admitted to concert
DEBUG: giving ticket to e (group z)
DEBUG: giving ticket to f (group z)
DEBUG: Concert:
cash: deque([Person(g,w,False)])
entrance: deque([Person(e,z,True),
Person(f,z,True)])
[80]:
['d']
[81]:
con.dequeue()
DEBUG: DEQUEUING ..
DEBUG: e (group z) admitted to concert
DEBUG: f (group z) admitted to concert
DEBUG: giving ticket to g (group w)
DEBUG: Concert:
cash: deque([])
entrance: deque([Person(g,w,True)])
[81]:
['e', 'f']
[82]:
con.dequeue()
DEBUG: DEQUEUING ..
DEBUG: g (group w) admitted to concert
DEBUG: Concert:
cash: deque([])
entrance: deque([])
[82]:
['g']
[83]:
# calling dequeue on empty lines gives empty list:
con.dequeue()
DEBUG: DEQUEUING ..
DEBUG: Concert:
cash: deque([])
entrance: deque([])
[83]:
[]
Special dequeue case: broken group
In the special case when there is a group at the entrance with one or more members without a ticket, it is assumed that the group gets broken, so whoever has the ticket enters and the others get enqueued at the cash.
[84]:
con = Concert()
con.enqe(Person('a','x',True))
con.enqe(Person('b','x',False))
con.enqe(Person('c','x',True))
con.enqc(Person('f','y',False))
con
[84]:
Concert:
cash: deque([Person(f,y,False)])
entrance: deque([Person(a,x,True),
Person(b,x,False),
Person(c,x,True)])
[85]:
con.dequeue()
DEBUG: DEQUEUING ..
DEBUG: a (group x) admitted to concert
DEBUG: b (group x) has no ticket! Sending to cash
DEBUG: c (group x) admitted to concert
DEBUG: giving ticket to f (group y)
DEBUG: Concert:
cash: deque([Person(b,x,False)])
entrance: deque([Person(f,y,True)])
[85]:
['a', 'c']
[86]:
con.dequeue()
DEBUG: DEQUEUING ..
DEBUG: f (group y) admitted to concert
DEBUG: giving ticket to b (group x)
DEBUG: Concert:
cash: deque([])
entrance: deque([Person(b,x,True)])
[86]:
['f']
[87]:
con.dequeue()
DEBUG: DEQUEUING ..
DEBUG: b (group x) admitted to concert
DEBUG: Concert:
cash: deque([])
entrance: deque([])
[87]:
['b']
[88]:
con
[88]:
Concert:
cash: deque([])
entrance: deque([])
8. OfficeQueue
An office offers services 'x', 'y' and 'z'. When people arrive at the office, they state which service they need, get a ticket and enqueue. Suppose at the beginning of the day we are considering there is only one queue.
The office knows on average how much time each service requires:
[89]:
SERVICES = { 'x':5, # minutes
'y':20,
'z':30
}
With this information it is able to inform new clients approximately how long they will need to wait.
OfficeQueue is implemented as a linked list, where people enter the queue from the tail and leave from the head. We can represent it like this (NOTE: ‘cumulative wait’ is not actually stored in the queue):
wait time: 155 minutes
cumulative wait: 5 10 15 45 50 55 85 105 110 130 150 155
wait times: 5 5 5 30 5 5 30 20 5 20 20 5
x x x z x x z y x y y x
a -> b -> c -> d -> e -> f -> g -> h -> i -> l -> m -> n
^ ^
| |
head tail
Each node holds the client identifier 'a', 'b', 'c', and the service label (like 'x') requested by the client:
class Node:
def __init__(self, initdata, service):
self._data = initdata
self._service = service
self._next = None
OfficeQueue keeps fields _services, _size and a field _wait_time which holds the total wait time of the queue:
class OfficeQueue:
def __init__(self, services):
self._head = None
self._tail = None
self._size = 0
self._wait_time = 0
self._services = dict(services)
[90]:
from office_queue_sol import *
SERVICES = { 'x':5, # minutes
'y':20,
'z':30
}
oq = OfficeQueue(SERVICES)
print(oq)
OfficeQueue:
[91]:
oq.enqueue('a','x')
oq.enqueue('b','x')
oq.enqueue('c','x')
oq.enqueue('d','z')
oq.enqueue('e','x')
oq.enqueue('f','x')
oq.enqueue('g','z')
oq.enqueue('h','y')
oq.enqueue('i','x')
oq.enqueue('l','y')
oq.enqueue('m','y')
oq.enqueue('n','x')
[92]:
print(oq)
OfficeQueue:
x x x z x x z y x y y x
a -> b -> c -> d -> e -> f -> g -> h -> i -> l -> m -> n
[93]:
oq.size()
[93]:
12
Total wait time can be accessed from outside with the method wait_time():
[94]:
oq.wait_time()
[94]:
155
ATTENTION: you only need to implement the methods time_to_service and split
DO NOT touch other methods.
8.1 - time_to_service
Open file office_queue_exercise.py with and start editing.
In order to schedule work and pauses, for each service office employees want to know after how long they will have to process the first client requiring that particular service.
First service encountered will always have a zero time interval (in this example it’s x):
wait time: 155
cumulative wait: 5 10 15 45 50 55 85 105 110 130 150 155
wait times: 5 5 5 30 5 5 30 20 5 20 20 5
x x x z x x z y x y y x
a -> b -> c -> d -> e -> f -> g -> h -> i -> l -> m -> n
|| | |
x : 0 | |
| | |
|---------------| |
| z : 15 |
| |
|-----------------------------------|
y : 85
[95]:
SERVICES = { 'x':5, # minutes
'y':20,
'z':30
}
oq = OfficeQueue(SERVICES)
print(oq)
OfficeQueue:
[96]:
oq.enqueue('a','x')
oq.enqueue('b','x')
oq.enqueue('c','x')
oq.enqueue('d','z')
oq.enqueue('e','x')
oq.enqueue('f','x')
oq.enqueue('g','z')
oq.enqueue('h','y')
oq.enqueue('i','x')
oq.enqueue('l','y')
oq.enqueue('m','y')
oq.enqueue('n','x')
print(oq)
OfficeQueue:
x x x z x x z y x y y x
a -> b -> c -> d -> e -> f -> g -> h -> i -> l -> m -> n
Method to implement will return a dictionary mapping each service to the time interval after which the service is first required:
[97]:
oq.time_to_service()
[97]:
{'x': 0, 'y': 85, 'z': 15}
Services not required by any client
As a special case, if a service is not required by any client, its time interval is set to the queue total wait time (because a client requiring that service might still show up in the future and get enqueued)
[98]:
oq = OfficeQueue(SERVICES)
oq.enqueue('a','x') # completed after 5 mins
oq.enqueue('b','y') # completed after 5 + 20 mins
print(oq)
OfficeQueue:
x y
a -> b
[99]:
print(oq.wait_time())
25
[100]:
oq.time_to_service() # note z is set to total wait time
[100]:
{'x': 0, 'y': 5, 'z': 25}
Now implement this:
def time_to_service(self):
""" RETURN a dictionary mapping each service to the time interval after which
the service is first required.
- the first service encountered will always have a zero time interval
- If a service is not required by any client, time interval is set to
the queue total wait time
- MUST run in O(n) where n is the size of the queue.
"""
Testing: python3 -m unittest office_queue_test.TestTimeToService
8.2 split
Suppose a new desk is opened: to reduce waiting times the office will comunicate on a screen to some people in the current queue to move to the new desk, thereby creating a new queue. The current queue will be split in two according to this criteria: after the cut, the total waiting time of the current queue should be the same or slightly bigger than the waiting time in the new queue:
ATTENTION: This example is different from previous one!
Total wait time here is 150 instead of 155
ORIGINAL QUEUE:
wait time = 150 minutes
wait time / 2 = 75 minutes
cumulative wait: 30 50 80 110 115 120 140 145 150
wait times: 30 20 30 30 5 5 20 5 5
z y z z x x y x x
a -> b -> c -> d -> e -> f -> g -> h -> i
^ ^ ^
| | |
head cut here tail
MODIFIED QUEUE:
wait time: 80 minutes
wait times: 30 20 30
cumulative wait: 30 50 80
z y z
a -> b -> c
^ ^
| |
head tail
NEW QUEUE:
wait time: 75 minutes
wait times: 30 5 5 20 5 5
cumulative wait: 30 35 40 60 65 70
z x x y x x
d -> e -> f -> g -> h -> i
^ ^
| |
head tail
Implement this method:
def split(self):
""" Perform two operations:
- MODIFY the queue by cutting it so that the wait time of this cut
will be half (or slightly more) of wait time for the whole original queue
- RETURN a NEW queue holding remaining nodes after the cut - the wait time of
new queue will be half (or slightly less) than original wait time
- If queue to split is empty or has only one element, modify nothing
and RETURN a NEW empty queue
- After the call, present queue wait time should be equal or slightly bigger
than returned queue.
- DO *NOT* create new nodes, just reuse existing ones
- REMEMBER to set _size, _wait_time, _tail in both original and new queue
- MUST execute in O(n) where n is the size of the queue
"""
Testing: python3 -m unittest office_queue_test.SplitTest
[ ]: