integer_programming_drivers.py

#!/usr/bin/env python

from collections import defaultdict
from pprint import pprint
import random

from pulp import *


def generate_trips(route_names, start_hour, end_hour):
  results = []
  for route_name in route_names:
    for hour in range(start_hour, end_hour):
      results.append((route_name, hour))
  return results


def generate_duties(num_duties, trips):
  # Create some data structues for each lookup of:
  # - The possible hours of the day
  # - The possible routes for each hour
  possible_hours = {t[1] for t in trips}
  possible_routes_for_hour = defaultdict(list)
  for t in trips:
    possible_routes_for_hour[t[1]].append(t[0])

  duties = []

  for _ in range(num_duties):
    num_trips = random.randrange(1, len(possible_hours) + 1)
    hours = random.sample(list(possible_hours), num_trips)
    duty = []
    for hour in sorted(hours):
      route_name = random.choice(possible_routes_for_hour[hour])
      duty.append((route_name, hour))
    duties.append(duty)

  return duties


def cost(duty):
  """
  Assume units of pay is "hours". I.e. for each hour worked increment by 1.

  Rules:
  1. Drivers get paid for each hour they work, but not the break.
  2. They get a minimum pay of eight hours for coming into work (known as guarantee pay).
  3. They get 1.5x pay for each hour they work over eight hours(overtime pay).
  4. Each duty over four hours should allow the driver to have a break.
  """

  # Rule 1 & 2
  pay = max(len(duty), 8)
  # Rule 3
  if len(duty) > 8:
    pay += 0.5 * (len(duty) - 8)

  # Rule 4: If there's no break, say pay to a very large value, since it violates a
  # basic labor law.
  if len(duty) > 4:
    first_hour = duty[0][1]
    last_hour = duty[-1][1]
    if last_hour - first_hour + 1 == len(duty):
      # No break
      pay = int(1e6)

  return pay


def solve(duties, trips):
  problem = LpProblem('driver_scheduling', LpMinimize)
  variables = []
  costs = []
  # Data structure to generate constraints for each trip.
  variables_for_trip = {trip: [] for trip in trips}

  # We have to make sure there's some set of duties that really do include all
  # the trips. Since we randomly generate duties, we can't be sure. E.g., we
  # could get unlucky and randomly generate duties that all only have trips on
  # the 'A' route. To solve this problem, generate one duty for each trip that
  # includes only that trip.
  duties = duties + [[trip] for trip in trips]

  # Gather up variables and costs
  for i, duty in enumerate(duties):
    # Create a binary variable (the value can be only 0 or 1)
    x = LpVariable('x{}'.format(i + 1), 0, 1, LpBinary)
    variables.append(x)
    costs.append(cost(duty))
    for trip in duty:
      variables_for_trip[trip].append(x)

  # Create the objective function. lpDot is shorthand for
  # c * x for (c, x) in zip(costs, variables)
  problem += lpDot(costs, variables)

  # Create constraints that for each trip, exactly one x from the duties
  # including it must be 1.
  for xs in variables_for_trip.values():
    problem += lpSum(xs) == 1

  # Pulp gives a very nice string representation of the problem when printed.
  print(problem)
  status = problem.solve()
  print(LpStatus[status])

  # We have a solution, now look at the values of xs to determine which duties
  # to use. Sum the cost for each used duty.
  solution = []
  total_cost = 0
  for i, x in enumerate(variables):
    if x.value():
      solution.append(duties[i])
      total_cost += costs[i]

  return solution, total_cost


def main():
  routes = 'ABCD'
  start_hour = 9
  end_hour = start_hour + 12

  trips = generate_trips(routes, start_hour, end_hour)
  duties = generate_duties(5, trips)
  solution_duties, solution_cost = solve(duties, trips)
  print("Cost: {}".format(solution_cost))
  pprint(solution_duties)


if __name__ == '__main__':
  main()