prodplan_ch.py

# Construction Heuristic for Production Planning Instances
# Author: Fabian Leuthold

import re
import numpy as np
import matplotlib.pyplot as plt

def read_ppl_file(file_path):
    """
    read a production planning problem to assign m jobs to n (identical) machines
    to minimize production time span

    :param file_path:
    :return:
    """
    with open(file_path, 'r') as file:
        num_machines = 0
        job_durations = []
        in_jobs_section = False

        for line in file:
            line = line.strip()
            if line.startswith("NUM_MACHINES"):
                num_machines = int(line.split(":")[1].strip())
            elif line.startswith("NUM_JOBS"):
                continue
            elif line.startswith("JOBS_ID_DURATION"):
                in_jobs_section = True
            elif in_jobs_section and line != "EOF":
                parts = line.split()
                if len(parts) == 2:
                    job_duration = float(parts[1])
                    job_durations.append(job_duration)
            elif line == "EOF":
                break

    return num_machines, job_durations


def production_greedy(num_machines, job_durations):
    """
    construction heuristic to generate a fair solution to the production problem and distribute the jobs over the
    (identical) machines in order to minimize overall production timespan. returns a dictionary with the machine-id as
    key and the list of assigned job-ids.
    it works as follows: all jobs are sorted by descending duration, then iteratively assigned to the machine having
    the shortest production time so far.

    :param num_machines: the number of (identical) machines which can process the jobs
    :param job_durations: the durations of each job in a list
    :return: a dictionary with the machine-id as key and the list of assigned job-ids.
    """

    # Sort the jobs by duration in descending order
    sorted_jobs = sorted(range(len(job_durations)), key=lambda x: job_durations[x], reverse=True)

    # Initialize a list to keep track of jobs assigned to each machine
    machine_jobs = [[] for _ in range(num_machines)]
    # Initialize a list to keep track of the total duration on each machine
    machine_loads = [0] * num_machines

    # Assign jobs to machines
    for job_id in sorted_jobs:
        duration = job_durations[job_id]
        # Find the machine with the minimum load
        min_machine = min(range(num_machines), key=lambda x: machine_loads[x])
        # Assign the job to this machine
        machine_jobs[min_machine].append(job_id)
        # Update the load of the machine
        machine_loads[min_machine] += duration

    return machine_jobs


def plot_job_assignments(solution, job_durations, instance_name):
    """
    plot the solution

    :param solution: the solution
    :param job_durations: the job durations
    :return:
    """
    num_machines = len(solution)
    colors = plt.cm.get_cmap('tab20', len(job_durations))
    fig, ax = plt.subplots(figsize=(6, 3), dpi=80)

    max_durations = []
    for machine_index, jobs in enumerate(solution):
        current_time = 0
        for job_index in jobs:
            job_duration = job_durations[job_index]
            ax.broken_barh([(current_time, job_duration)],
                           (machine_index - 0.4, 0.8),
                           facecolors=colors(job_index),
                           edgecolor='black',
                           linewidth=1)
            # Textbeschriftung hinzufügen
            ax.text(current_time + job_duration / 2, machine_index,
                    f'J{job_index + 1}\n{job_duration}',
                    ha='center', va='center', color='white', fontsize=12, fontweight='bold')
            current_time += job_duration
            max_durations.append(current_time)

    max_duration = round(max(max_durations), 1)

    ax.axvline(max_duration, color='red', linestyle='--', linewidth=2)

    # set axes and labels
    ax.set_yticks(np.arange(num_machines))
    ax.set_yticklabels([f"M{i + 1}" for i in range(num_machines)], fontsize=12, fontweight='bold')
    ax.set_xlabel('Time', fontsize=12, fontweight='bold')
    ax.set_title("Job Assignments for '" + instance_name + "'", fontsize=14, fontweight='bold')
    ax.grid(True, linestyle='--', alpha=0.7)
    ax.tick_params(axis='y', which='both', length=0)  # Entfernen der Y-Achsen-Ticks

    # add max duration to the x-ticks
    x_ticks = np.append(ax.get_xticks(), max_duration)
    # only keep positive x-ticks
    x_ticks = [tick for tick in x_ticks if tick >= 0.0]
    ax.set_xticks(x_ticks)
    ax.set_xticklabels([f'{int(tick)}' if tick != max_duration else f'{tick}' for tick in x_ticks], fontsize=12)

    plt.tight_layout()
    plt.show()


# *******************************************************************************************************************
# * Testcases to validate if code is working as expected                                                            *
# *******************************************************************************************************************


def test_read_test():
    """
    test, whether reading function for PRODPLAN_M2_J5 provides the expected output

    :return:
    """
    # read problem instance
    file_path = './data/PRODPLAN_M2_J5.ppl'

    num_machines, job_durations = read_ppl_file(file_path)
    assert num_machines == 2, "expected 2 machines but found else."
    assert job_durations == [1.8, 8.6, 7.8, 2.9, 5.2], "expected [1.8, 8.6, 7.8, 2.9, 5.2] durations but found else."


def test_kh_test_1():
    """
    test, whether solution for PRODPLAN_M2_J4_test upholds the expected solution

    :return:
    """
    # read problem instance
    file_path = './data/PRODPLAN_M2_J4_test.ppl'

    num_machines, job_durations = read_ppl_file(file_path)
    machine_job_dict = production_greedy(num_machines, job_durations)
    assert num_machines == 2, "expected 2 machines but found else."
    assert machine_job_dict[0] == [1], "expected job [1] assigned to machine 1 but found else."
    assert machine_job_dict[1] == [0, 2, 3], "expected jobs [0, 2, 3] assigned to machine 2 but found else."


def test_kh_test_2():
    """
    test, whether solution for PRODPLAN_M2_J4_test2 upholds the expected solution

    :return:
    """
    # read problem instance
    file_path = 'data/PRODPLAN_M2_J5_test2.ppl'

    num_machines, job_durations = read_ppl_file(file_path)
    machine_job_dict = production_greedy(num_machines, job_durations)
    assert num_machines == 2, "expected 2 machines but found else."
    assert machine_job_dict[0] == [0, 3, 4], "expected job [0, 3, 4] assigned to machine 1 but found else."
    assert machine_job_dict[1] == [1, 2], "expected jobs [1, 2] assigned to machine 2 but found else."


# *******************************************************************************************************************
# * Main Program                                                                                                    *
# *******************************************************************************************************************


def main():
    file_paths = ['./data/PRODPLAN_M2_J5.ppl', './data/PRODPLAN_M2_J8.ppl', './data/PRODPLAN_M2_J21.ppl',
                  './data/PRODPLAN_M4_J32.ppl', './data/PRODPLAN_M2_J4_test.ppl', './data/PRODPLAN_M2_J5_test2.ppl']

    for id, file in enumerate(file_paths):
        # Read the problem instance
        num_machines, job_durations = read_ppl_file(file)
        # Create a first solution applying a primitive k-heuristic
        machine_jobs = production_greedy(num_machines, job_durations)
        # Calculate the total load for each machine using list comprehension
        machine_loads = [sum(job_durations[job_id] for job_id in jobs) for jobs in machine_jobs]

        print(f"*** File {file}:")
        print(f"Num Machines: {num_machines}")
        print(f"Num Jobs: {len(job_durations)}")
        print(f"Solution:")
        print(f"Production Time-Span: {round(max(machine_loads), 1)}")
        for machine_id, job_ids in enumerate(machine_jobs):
            # Convert job_ids to a comma-separated string and print with machine_id
            job_list = ", ".join(map(str, [job_id + 1 for job_id in job_ids]))
            print(f"Machine-ID {machine_id + 1}: {job_list}")
        # plot result
        instance_name_match = re.search(r'([^/]+)\.ppl$', file)
        if instance_name_match:
            instance_name = instance_name_match.group(1)
        else:
            instance_name = "n/a"
        plot_job_assignments(machine_jobs, job_durations, instance_name)


if __name__ == "__main__":
    main()