How to use the framework¶
The Business Process Optimization framework can be used to simulate an operational business process in combination with a particular technique for forecasting and a particular technique for planning. The results of different variants (of the process, forecasting technique or planning technique) can be compared and visualized.
To optimize a business process, three elements must be specified:
the business process itself, using a problem description.
a planner that decides which resource to assign to which task at what moment in time, possibly based on predictions.
optionally, a predicter that either forecasts when cases will arrive to the process, or predicts how long a task in the process will take, or both.
The planner and predicter can be freely defined, but some default planners and predicters also exist.
Contents
Quick start¶
The code below illustrates how to quickly get started with an example problem.
from problems import MMcProblem
from simulator import Simulator, Reporter
from planners import GreedyPlanner
problem = MMcProblem()
planner = GreedyPlanner()
reporter = Reporter(10000)
results = Simulator.replicate(problem, planner, reporter, 50000, 10)
print(Reporter.aggregate(results))
The example problem represents an M/M/c queue with an interarrival rate of 10 a service time of 9 and c=2 servers. The specification of the problem can be inspected in the MMcProblem code.
First, the problem must be created. The problem will generate cases that arrive at a simulated time. Each case has one or more tasks that must be performed on it (in this case only one task).
Second, a planner should be created that plans the assignment of tasks to resources. A default greedy planner exists that assigns each task to the first available resource.
Third, a reporter should be created that collects information from the simulator. A default reporter exists that collects for each instance the number of cases that completed and the average cycle time per case as well as the number of tasks that completed and the average processing time and waiting time per task. The reporter takes the warmup time as a parameter, i.e. the simulation time during which no data is collected for reporting.
Now, the simulator can be run for a number of replications of the problem with the planner and the reporter. The simulator also takes an amount of simulation time for which it should be run. The simulator returns a list of results. This is the information that the reporter generates for each instance, i.e. the list has as many results as there are problem instances.
Finally, the list of results can be aggregated. Each reporter has an aggregate method that aggregated the results over all problem instances into a single result. The default reporter calculates the averages and 95% confidence intervals over all the reported statistics: the number of cases, etc.
Defining a problem¶
A problem can be defined as an subclass of the abstract class problems.Problem.
By doing so it inherits methods for generating problem instances and loading and saving problem instances.
A problem must define:
The resources that exist.
The interarrival time distribution of cases.
The types of tasks that can be performed for cases.
Rules for what the next task will be when a case first arrives or when a task is completed.
The processing time distribution for each task.
The data that is generated by a task.
For each task the list of resources that is authorized to perform the task.
The resources and task types are static lists, typically of labels for resources and task types. For example, we can define the M/M/c problem presented above with two resources and a single task type as follows.
class MMcProblem(Problem):
resources = ["R1", "R2"]
task_types = ["T"]
The other elements of the problem are implemented as methods that sample from a distribution. For example, in case of the M/M/c problem, there is a method that samples the processing time from an exponential distribution with a mean of 10. It must the sampled value. There are also methods that sample the next type of task that must be performed. However, for an M/M/c queue this is quite simple, because it must always be T as the first task type and no tasks after T is performed. However, it is also possible to simulate a queuing network, an MDP or a business process model. In those cases the next task would vary.
More precisely, the remainder of the M/M/c problem can be defined as follows.
def processing_time_sample(self, resource, task):
return random.expovariate(1/9)
def interarrival_time_sample(self):
return random.expovariate(1/10)
def sample_initial_task_type(self):
return "T"
def next_task_types_sample(self, task):
return []
def resource_pool(self, task_type):
return self.resources
def data_sample(self, task_type):
return dict()
This samples the processing time and the interarrival time from exponential distributions. It samples the initial task type as T and the subsequent task types as empty, explained above. It specifies that it always returns the list of all resources as the resources that are authorized to perform a task and it always returns an empty dictionary as the data that is generated by a task, i.e. there is no data.
Defining a planner¶
A planner can be defined as a subclass of the abstract class planners.Planner.
A planner is called each time a new task or a resource becomes available in the simulator.
It must then assign resources to tasks and return which resource to assign to which task.
To decide on the assignment, the planner has access to the current state of the cases that
are being simulated via the environment that is passed to it. Most importantly it has access to:
assigned_tasks: The tasks that are currently assigned.
unassigned_tasks: The tasks that are currently not assigned.
available_resources: The set of resources that are currently available.
busy_resources: The resources that are currently busy.
reserved_resources: The resources that are currently reserved.
busy_cases: The cases of which a task is currently being performed.
now: The current simulation time.
It can use all of this information to decide which resource to assign to which task.
The lifecycle of tasks and resources is important. Initially, all tasks are unassigned and all resources are available. the list of busy cases identifies the cases and the next tasks to perform for those cases. When the planner assigns a resource to a task, it also passes the time at which the resource must start executing the task. The resource then becomes reserved and the task becomes assigned. At the moment the resource starts executing the task, the resource becomes busy. When the resource is done executing the task, the task is removed from both the list of assigned and the list of unassigned tasks and the resource becomes available again. At that moment the task is also removed from the busy cases. The simulator then calculates the next tasks that must be performed, which are added to the unassigned tasks and the busy cases.
class GreedyPlanner(Planner):
def assign(self, environment):
assignments = []
available_resources = environment.available_resources.copy()
for task in environment.unassigned_tasks.values():
if len(available_resources) > 0:
resource = available_resources.pop()
assignments.append((task, resource, environment.now))
return assignments
This planner simply iterates over all unassigned tasks and assigns the first available resource to it. It does this by adding a triple - task, resource, moment of assignment - to the list of assignments, which is returned in the end. Note that unassigned_tasks is a dictionary that maps a task identifier to a task. This is the reason the function iterates over unassigned_tasks.values().
Defining a predicter¶
A predicter can be defined as a subclass of the abstract class predicters.Predicter.
It can be passed to a planner to use the prediction when making a plan.
A predicter can implement methods for calculating:
the time a resource will take to perform a particular task.
the remaining time a resource will still take to perform a task to which it is assigned.
the next task that will arrive.
For example, let’s create a new variant of the M/M/c problem in which each task is about one of two kinds of applications (uniformly distributed) and resource R1 is better at processing one kind of application, while resource R2 is better at processing the other. Specifically, resources take 9 minutes (exponentially distributed) on the applications they are good at, while they take 27 minutes (exponentially distributed) on the applications that they are not good at.
The code below shows how the original M/M/c problem can be adapted to represent that kind of behavior. Most importantly, each task now has a data element optimal_resource that identifies the resource that is good at the task.
class ImbalancedProblem(Problem):
...
def processing_time_sample(self, resource, task):
if resource == task.data["optimal_resource"]:
return random.expovariate(1/9)
else:
return random.expovariate(1/27)
def data_sample(self, task_type):
data = dict()
data["optimal_resource"] = random.choice(["R1", "R2"])
return data
Accordingly, we can create a predicter that predicts the processing time and the remaining processing time of a task, simply as the average processing time of the resource processing the task as follows.
class ImbalancedPredicter(Predicter):
@staticmethod
def predict_processing_time_task(problem, resource, task):
if resource == task.data["optimal_resource"]:
return 9
else:
return 27
@staticmethod
def predict_remaining_processing_time(problem, resource, task, start_time, now):
return ImbalancedPredicter.predict_processing_time_task(problem, resource, task)
Now we can create an alternative planner that does not try to assign a task to the first available resource, but rather to the resource that is expected to complete the task fastest, based on the prediction.
Visualizing the results¶
There are some convenience functions for visualizing the results of a simulation. These are specifically created to work well with the reporters.
The default reporter generates a dictionary with the performance indicators as keys and
lists as values, where each element is the result of the performance indicator for a replication.
For example, if you do 3 replications, the code below could print something like [3987, 4010, 3996].
...
results = Simulator.replicate(problem_instances, planner, reporter, 50000)
print(results["cases completed"])
We can pass this on to the boxplot visualizer, which takes a dictionary, where the keys are the labels of the boxplots and the values are the lists on the basis of which each boxplot is generated. For example, we can generate a boxplot for the code above as follows:
visualizers.boxplot({'my boxplot title': results["cases completed"]})
The default reporter’s aggregator aggregates the individual values into a pair of an average and a 95% confidence interval. We can pass this on the lineplot visualizer, which visualizes it as a line of averages, with lines representing the 95% confidence intervals above and below that line.
For example, suppose we have created problems for processing times of 4 to 9 minutes, we can simulate and create a lineplot visualization for those problems as follows:
cases_completed_for_processing_time = dict()
for processing_time in range(4, 10):
... # create problem_instances for that processing_time
results = Simulator.replicate(problem_instances, planner, reporter, 50000)
aggregated_results = Reporter.aggregate(results)
cases_completed_for_processing_time[processing_time] = aggregated_results["cases completed"]
line_with_ci(cases_completed_for_processing_time)