# Espresso Queue simulation

I was asked to do a technical test with the following specification, but my solution was rejected and I was not selected for the available position (Junior to "normal" level, with 4 days of time to finish the test). Could you point out which areas need improvement in my solution?

In software company X, engineers work best when consuming one cup of espresso an hour. The office espresso machine has a first-come-first-serve queue that applies to everyone, except for certain "super busy" engineers who are prioritized before non-super-busy ones. Among competing super-busies the first-come-first-serve principle applies again.

Please implement a simulator for this espresso machine. Input parameters are number of engineers, the chance that an engineer becomes super-busy in some unit of time, and for how long they stay super-busy.

And my solution in Python 3.4: (GitHub version)

machine.py

import msvcrt # Windows-only non-blocking input. Universal would need a lot more things
import random
from classes.espresso import Espresso
from classes.queue import Queue
from classes.coffeeenum import CoffeeEnum
from classes.order import Order

# Launch the machine
num_engineers = 1705
chance_superbusy = 25 # percent
max_time_superbusy = 120 # minutes
espresso_machine = Espresso(num_engineers, chance_superbusy, max_time_superbusy)
espresso_machine.start()

# Setup a few things
queue = Queue()
queue_id = 0
coffee_name = CoffeeEnum()
superbusy = False

# Run the show!
espresso_machine.display_coffee_choices()
while True:
espresso_machine.brew(queue.get_next(), queue)

# Read input to get new orders and enter superbusy mode
if msvcrt.kbhit():
# There's a chance the engineer is superbusy!
if random.randrange(0, 100) < chance_superbusy:
print("By surprise, you are now superbusy!")
superbusy = True

input = msvcrt.getch().decode("ascii").upper()
# Engineer can also choose to be superbusy to skip the line (hmm...)
if input == "Y":
print("You are superbusy! Your order now has priority.")
superbusy = True
if input == "S":
queue.status()
if input.isdigit():
if int(input) in range(1, 6):
print("You ordered:", coffee_name.get_name(input))
queue_id += 1
new_order = [queue_id, input, superbusy]
queue.append(new_order)
superbusy = False


classes/queue.py:

from classes.coffeeenum import CoffeeEnum
from classes.names import Names

class Queue:
normal_queue = []
super_queue = []
# Gets names of coffee types
coffee_enum = CoffeeEnum()
# Connect queue ID with a person name
names = Names()

def __init__(self):
print("----------")

def get_next(self):
if self.super_queue:
return self.super_queue[0]
elif self.normal_queue:
return self.normal_queue[0]

def has_orders(self):
if self.super_queue or self.normal_queue:
return True
else:
return False

def status(self):
if self.has_orders():
print("Queue contains:")
print("=====")
if self.super_queue:
print("> Superbusy orders:")
for order in self.super_queue:
print(self.names.get_name(order[0]),
"ordered:", self.coffee_enum.get_name(order[1]))
if self.normal_queue:
print("> Normal orders:")
for order in self.normal_queue:
print(self.names.get_name(order[0]),
"ordered:", self.coffee_enum.get_name(order[1]))
print("=====")
else:
print("Queue is empty! Ready for orders...")

def append(self, list_order):
self.coffee_enum.get_name(list_order[1]),
"for", self.names.get_name(list_order[0]),
"(Superbusy:", list_order[2], ")")
if list_order[2]: # Superbusy
self.super_queue.append(list_order)
else:
self.normal_queue.append(list_order)

def remove_order(self, list_order):
if list_order:
if list_order[2]: # Superbusy order
self.super_queue.remove(list_order)
else:
self.normal_queue.remove(list_order)


classes/espresso.py:

from time import time
from classes.queue import Queue

class Espresso:
num_engineers = 0
chance_superbusy = 0 # percent
max_time_superbusy = 0 # minutes
is_running = False
is_brewing = False
brew_started_time = 0
brew_ready_time = time() + 2 # +2 to prevent unneeded message on launch
queue = 0
current_brew = 0

def __init__(self, num_engineers, chance_superbusy, max_time_superbusy):
print("Espresso is booting...")
print("> Engineer count:", num_engineers)
print("> Chance superbusy:", chance_superbusy, "%")
print("> Max time being superbusy:", max_time_superbusy, "minutes")

def start(self):
is_running = True
print("Espresso machine started OK")
print("> Press any key except A-Z or 0-9 to exit")

def display_coffee_choices(self):
print("Pick your poison, Mr. Engineer! \n"
"If you are superbusy, press (Y) for priority lane! \n"
"(S) for Queue Status")
print("1 - Espresso")
print("2 - Cappuccino")
print("3 - Latte")
print("4 - Macchiato")
print("5 - Irish Coffee")
return "" # Otherwise will display "None"

if time() >= self.brew_ready_time and self.is_brewing:
self.queue.remove_order(self.current_brew)
if not self.queue.has_orders():
print("Queue empty, ready for new orders!")
self.is_brewing = False
if self.is_brewing:
return False
else:
return True

def brew(self, coffee, queue):
if queue.has_orders():
self.queue = queue
self.current_brew = coffee
print("Brewing now... ready in 15 seconds!")
self.brew_started_time = time()
self.is_brewing = True


classes/order.py:

from classes.coffeeenum import CoffeeEnum

class Order:
id = 0
product = 0
is_superbusy = False

def __init__(self, list_order):
# This is used to get the coffee names from the id
coff = CoffeeEnum()

# Order comes in as a list: [id, product, boolean superbusy]
self.id = list_order[0]
self.product = coff.get_name(list_order[1])
self.is_superbusy = list_order[2]


classes/names.py:

import random

class Names:
def get_name(self, id):
id = int(id)
# Keep value in 0-9 range for demo purposes
if id >= 10:
id = random.randint(0, 9)
return {
1: "Albert",
2: "Benjamin",
3: "Charles",
4: "Dmitri",
5: "Euclid",
6: "Francis",
7: "Gustav",
8: "Hermann",
9: "Isaac",
0: "John",
}.get(id, "Bjarne")


classes/names.py:

class CoffeeEnum:
def get_name(self, id):
id = int(id)
return {
1: "Espresso",
2: "Cappuccino",
3: "Latte",
4: "Macchiato",
5: "Irish Coffee",
}.get(id, "Coffee")

• Btw, here's how much code you would need to get non-blocking input in a cross-platform way (you will see why I left it out, heh): repolinux.wordpress.com/2012/10/09/… Commented Sep 11, 2014 at 10:27
• Was there any more to the question, eg how long does it take to make a coffee? Do engineers enter the queue on the hour, every hour, or at a random time during each hour? Or do you read a list of engineer IDs/enter queue times? Etc. Commented Sep 11, 2014 at 22:23
• All the information that was given is in the orange/pink box above. Commented Sep 12, 2014 at 11:38

### 1. Failed requirements

The program fails to implement the following details from the specification:

1. The specification says, "Input parameters are ..." but these parameters as implemented as global variables and not as inputs.

2. One of the parameters is "number of engineers", but the program makes no use of this number; it does not even represent the engineers at all.

3. Another parameter is "the chance that an engineer becomes super-busy in some unit of time" (my emphasis). But what the program implements is the chance that the current engineer becomes super-busy when a key is pressed.

4. The third parameter is "how long they stay super-busy" but in this implementation an engineer stops being super-busy as soon as they have placed an order.

### 2. Other problems

1. There's no documentation.

2. It looks to me as though the specification is looking for a simulator in the sense of "computer model of a phenomenon" — a program which we could use to investigate questions like, "how does the typical queue length vary as we change the number of engineers?" But this implementation is an interactive simulation, which might not be so useful, because someone has to babysit it in order to get any results.

3. The program fails to abstract away the measurement of time. This means that if some event takes 15 seconds in the simulation then we have to wait 15 seconds in real life for it to happen. This means that we can't exploit the power of computers to simulate events thousands or millions of times faster than they happen.

4. The program contains lots of features that weren't in the specification, like the names of the engineers, the different types of coffee, the queue status report, and so on. If the specification had been fully implemented, this kind of extra detail can be fun, but in the circumstances it looks like a distraction. (Also, given that you picked names for the engineers, did they really all have to be male names?)

5. The program only runs on Windows!

6. There are lots of minor problems that indicate a lack of care and attention. There's an Order class, but it is never used; the Espresso class maintains is_running and brew_started_time members but these are not used; Espresso.__init__ takes parameters num_engineers, chance_superbusy, max_time_superbusy but these are just printed out and not otherwise used; the module classs/espresso.py imports Queue but does not use it.

Specifications rarely provide a complete description of a program. This means that it is nearly always useful to write down a design that lists and justifies the decisions you have made. Here are mine:

1. The specification omits some necessary parameters to the simulation. We need to add parameters for how long an engineer works before taking a coffee break and how long an espresso takes to brew.

2. I've chosen to abstract away the notion of time passing in the form of a Schedule class. This processes events in order by their timestamp, which is just a number. Justification: we don't have to deal with the complexity of operating system real-time interfaces; also, we can run the simulation at whatever speed we like.

3. The scheduler stores future events in a priority queue implemented as a min-heap using the heapq module. Justification: can add events and retrieve the earliest event in logarithmic time.

4. I've chosen to represent an event as a function of one argument (the time at which it happens). Justification: separation of concerns (the scheduler doesn't need to have any knowledge about the events); easily extensible with more types of event.

5. I've implemented the queue for the espresso machine as a map from engineer to the time they joined the queue. The priority of super-busy engineers over ordinarily busy engineers is implemented by making the engineers orderable by giving them a __lt__ method. Justification: this ensures that we can't make a mistake that leads to the engineer being on the queue multiple times; it also avoids the complexity that would result if I had two queues (an engineer might change super-busy status while in the queue and this would require them to change queue).

6. I've represented parameters like "how long an engineer works before taking a coffee break" in the form of random variables, implemented as functions of zero arguments returning a duration. Justification: real-world processes have variation in how long they take; the implementation gives us the flexibility to easily change the way durations vary in case we've misunderstood the simulation requirements.

7. The specification says, "chance that an engineer becomes super-busy in some unit of time". I understand this to mean that the time an engineer remains (ordinarily) busy is exponentially distributed, and so I've implemented it using random.expovariate.

Here's the implementation:

from collections.abc import Iterator
import heapq
from itertools import count
from random import expovariate

class Schedule(Iterator):
"""An iterator yielding future events in order by time.

An event is represented by a function taking one argument (the
time at which the event happens). Schedule an event by calling the

Getting the next item from the iterator causes the earliest
remaining event to be called.

"""
def __init__(self):
self._events = [] # Priority queue in the form of a min-heap
self._seq = count() # Iterator yielding unique sequence ids

def __next__(self):
if not self._events:
raise StopIteration
time, _, event = heapq.heappop(self._events)
return time, event(time)

# Priority queue entries must be orderable, but functions are
# unorderable, so use a unique sequence number to break ties.
# See https://docs.python.org/3/library/heapq.html
heapq.heappush(self._events, (time, next(self._seq), event))

class Machine:
"""Model of an espresso machine with a queue of engineers. The
constructor takes arguments:

schedule -- a Schedule object managing the sequence of events
brew_variate -- function returning time taken to brew an espresso

"""
def __init__(self, schedule, brew_variate):
self.schedule = schedule
self.brew_variate = brew_variate
self.brewing = False
self._queue = dict()    # Map from engineer to time they joined queue

def enqueue(self, time, engineer):
"""An engineer joins the queue for the espresso machine at time."""
self._queue[engineer] = time
self.brew(time)

def brew(self, time):
"""Start brewing espresso at time (unless already brewing)."""
if not self.brewing and self._queue:
self.brewing = True

def serve(self, time):
"""Serve espresso to the engineer at the head of the queue at time."""
engineer, _ = min(self._queue.items())
del self._queue[engineer]
engineer.work(time)
self.brewing = False
self.brew(time)
return ('Espresso served to {} ({} waiting)'
.format(engineer, len(self._queue)))

class Engineer:
"""Model of an engineer. The constructor takes arguments:

schedule -- a Schedule object managing the sequence of events
machine -- a Machine object representing the espresso machine
time -- the time at which the engineer is created
work_variate -- function returning time to work
busy_variate -- function returning time to stay "busy"
superbusy_variate -- function return time to stay "super-busy"

"""
_ids = count()              # Iterator yielding engineer ids.

def __init__(self, schedule, machine, time, work_variate, busy_variate,
superbusy_variate):
self.id = next(self._ids)
self.schedule = schedule
self.machine = machine
self.work_variate = work_variate
self.busy_variate = busy_variate
self.superbusy_variate = superbusy_variate
self.busy(time)
self.work(time)

def __str__(self):
return 'Engineer {}'.format(self.id)

def __lt__(self, other):
# Super-busy engineers are more important (come earlier in
# queues) than ordinarily busy engineers.
return self.is_superbusy and not other.is_superbusy

def work(self, time):
"""Start work at time."""

def coffee_break(self, time):
"""Join the queue for the espresso machine at time."""
self.machine.enqueue(time, self)
return '{} went on a coffee break'.format(self)

def busy(self, time):
"""Become (ordinarily) busy at time."""
self.is_superbusy = False
return '{} stopped being super-busy'.format(self)

def superbusy(self, time):
"""Become super-busy at time."""
self.is_superbusy = True
return '{} became super-busy'.format(self)

def simulate(events=range(100),
engineers=20,
work_time=60,
superbusy_chance=0.01,
superbusy_time=120,
brew_time=3):
"""Run the espresso machine simulation. Parameters:

events -- iterable that terminates the simulation when it runs out.
engineers -- number of engineers
work_time -- mean time that an engineer works before going for espresso
superbusy_chance -- chance that engineer becomes super-busy per unit time
superbusy_time -- time that an engineer is super-busy
brew_time -- time taken to brew an espresso

"""
schedule = Schedule()
machine = Machine(schedule, lambda:brew_time)
for _ in range(engineers):
Engineer(schedule, machine, 0,
lambda:expovariate(1.0 / work_time),
lambda:expovariate(superbusy_chance),
lambda:superbusy_time)
for _, (time, result) in zip(events, schedule):
print("{:.2f}: {}".format(time, result))

• My understanding of the spec is that the program is supposed to keep track of the superbusy status of each engineer and change the status back to normal after certain time has elapsed. All your code does with the max_time_superbusy parameter is print it out.
• The spec mentions input parameters. Your program should input them somehow, eg. from the command line, instead of having them hardcoded in code.
• You are misusing class attributes where instance attributes would be appropriate. Creating multiple Queue instances would lead to surprising behavior as the instances would share the normal_queue and super_queue attributes.