Gantt Chart生成代码问题：并行流程未按预期执行求助

问题背景 #

我编写了一段生成甘特图的代码，当前接收的参数如下：

• 产品顺序：P_1,P_3,P_2
• 需求：['Glue Pick Place Screw', 'Screw Pick Glue Place', 'Screw Glue Place Pick']
• 分配的机器：[['M_1', 'M_3', 'M_4', 'M_2'], ['M_2', 'M_3', 'M_1', 'M_4'], ['M_2', 'M_1', 'M_4', 'M_3']]
• 机器耗时：{'M_1': 30, 'M_4': 20, 'M_3': 10, 'M_2': 30}

预期效果是遵循产品工序顺序的前提下优化时间，比如P_1用M_1启动，P_3用M_2启动，二者并行执行，但当前图表会先完成P_1的所有工序后，才开始处理P_3和P_2的并行流程，异常效果如下：

问题代码 #

# -*- coding: utf-8 -*-
"""
Created on Fri May 24 17:02:38 2024

@author: rafae
"""

import paho.mqtt.client as mqtt
import json
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches

# Define MQTT parameters
mqtt_broker = "localhost"
mqtt_port = 1883
mqtt_topic = "submodels"
product_json = []
machine_json = []
product_order = []
requirements = []
machine_assignment = []
# Define MQTT callback functions


def on_connect(client, userdata, flags, rc):
    print("Connected with result code " + str(rc))
    client.subscribe(mqtt_topic)
    client.subscribe("product_orders")


def on_message(client, userdata, msg):
    global product_json, machine_json, product_order, requirements, machine_assignment
    if msg.topic == "product_orders":
        try:
            product_order = msg.payload.decode()
            print("Produtos pedidos:")
            print(product_order)
        except Exception as e:
            print(f"Error processing product order message: {e}")
    if msg.topic == mqtt_topic:
        try:
            # Convert the received JSON string to a Python dictionary
            json_data = json.loads(msg.payload)
            # Process the JSON data
            machine_json, product_json = separate_json_data(json_data)
            # Ensure machine_json is a list of dictionaries
            if not isinstance(machine_json, list):
                raise ValueError("machine_json is not a list")
        except json.JSONDecodeError as e:
            print("Error decoding JSON:", e)
        except ValueError as ve:
            print("Error processing JSON data:", ve)
        except Exception as e:
            print(f"Unexpected error: {e}")
     
    if product_order:
        try:
            requirements = process_product_ids_from_string(product_json, product_order)
            print("Requirements")
            print(requirements)
            # print(requirements[0])
        except Exception as e:
            print(f"Error processing product IDs: {e}")

    if requirements:
        try:
            machine_assignment = assign_machines(requirements)
            print("Machines Assigned:")
            print(machine_assignment)
        except Exception as e:
            print(f"Error assigning machines: {e}")
   
    machine_time= extract_processing_times(machine_json)
    print(machine_time)
    
    if machine_assignment:
        try:
            generate_gantt_chart(machine_assignment, machine_time, product_order)
        except Exception as e:
            print(f"Error Generating Gantt Chart: {e}")
    
#separa o json em dois (um para as máquinas e outro para os produtos)
def separate_json_data(json_data):
    machines = []
    products = []

    for item in json_data:
        if 'assetAdministrationShells' in item:
            for asset_shell in item['assetAdministrationShells']:
                id_short = asset_shell['idShort']
                if 'Machine' in id_short:
                    machines.append(item)
                elif 'Product' in id_short:
                    products.append(item)

    return machines, products

#vai buscar os requirements de um produto através do id
def get_product_requirements(product_json, product_id):
    for product in product_json:
        for asset_shell in product.get('assetAdministrationShells', []):
            if asset_shell.get('id') == product_id:
                submodels = product.get('submodels', [])
                for submodel in submodels:
                    submodel_elements = submodel.get('submodelElements', [])
                    product_requirements = [element.get('value', '') for element in submodel_elements if element.get(
                        'idShort') == 'ProductRequirements']
                    return product_requirements

#separa a string de product order em vários ids
def process_product_ids_from_string(product_json, ids_string):
    if not ids_string:
        return []

    # Split the string by commas
    product_ids = ids_string.split(',')
    # Strip any whitespace or quotes from the IDs
    product_ids = [pid.strip() for pid in product_ids]

    all_requirements = []

    for product_id in product_ids:
        requirements = get_product_requirements(product_json, product_id)
        if requirements:
            all_requirements.append(','.join(requirements))

    return all_requirements

def assign_machines(requirements):
    machine_choices = []
    for req in requirements:
        assigned_machines = []
        individual_reqs = req.split(' ')
        for individual_req in individual_reqs:
            machine_id = find_machine_for_requirement(machine_json, individual_req.strip())
            if machine_id:
                assigned_machines.append(machine_id)
            else:
                print(f"No machine found for requirement: {individual_req.strip()}")
        machine_choices.append(assigned_machines)
    return machine_choices

def find_machine_for_requirement(machine_json, requirement):
    for machine in machine_json:
        for asset_shell in machine.get('assetAdministrationShells', []):
            submodels = machine.get('submodels', [])
            for submodel in submodels:
                submodel_elements = submodel.get('submodelElements', [])
                for element in submodel_elements:
                    if element.get('idShort') == 'MachineFunctions' and requirement in element.get('value', ''):
                        return asset_shell.get('id')
    return None

# Extract processing times from machine JSON
def extract_processing_times(machine_json):
    processing_times = {}
    for machine in machine_json:
        for asset_shell in machine.get('assetAdministrationShells', []):
            machine_id = asset_shell.get('id')
            for submodel in machine.get('submodels', []):
                for element in submodel.get('submodelElements', []):
                    if element.get('idShort') == 'MachineTime':
                        processing_times[machine_id] = int(element.get('value'))
    
    return processing_times


def generate_gantt_chart(machine_assignment, processing_times, product_order):
    fig, ax = plt.subplots(figsize=(10, 6))
    colors = ['blue', 'green', 'red', 'purple', 'orange', 'cyan']
    product_colors = {}
    start_times = {}
    product_labels = product_order.split(',')  # Split the product order string

    for idx, product in enumerate(product_labels):
        start_times[product] = 0
        product_colors[product] = colors[idx % len(colors)]

    machine_usage = {machine: 0 for machine in processing_times}

    for product_idx, machines in enumerate(machine_assignment):
        product = product_labels[product_idx]
        for machine in machines:
            start_time = max(start_times[product], machine_usage[machine])
            process_time = processing_times[machine]
            ax.barh(machine, process_time, left=start_time, color=product_colors[product], edgecolor='black')
            start_times[product] = start_time + process_time
            machine_usage[machine] = start_time + process_time

    # Add legend
    patches = [mpatches.Patch(color=color, label=product) for product, color in product_colors.items()]
    plt.legend(handles=patches)

    # Add labels and title
    ax.set_xlabel('Time')
    ax.set_ylabel('Machines')
    ax.set_title('Dynamic Gantt Chart')
    plt.grid(True)
    plt.show()

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

# Connect to MQTT broker
client.connect(mqtt_broker, mqtt_port, 60)

# Start the MQTT client loop
client.loop_forever()

问题原因 #

问题出在generate_gantt_chart函数的工序调度逻辑上：
原代码是按产品顺序逐个处理所有工序，也就是先把P_1的4个工序从头到尾执行完，才开始处理P_3的工序，完全没有考虑不同产品的工序可以在机器空闲时并行启动。

具体来说，原代码的嵌套循环会先遍历完第一个产品的所有工序，再进入第二个产品的工序处理，自然无法实现并行。

修正方案 #

改成按就绪状态调度工序：跟踪每个产品的当前待执行工序，以及每个机器的空闲时间，每次选择所有就绪的工序（产品的前序工序已完成，且对应机器空闲）来执行，直到所有工序完成。

修正后的generate_gantt_chart函数如下：

def generate_gantt_chart(machine_assignment, processing_times, product_order):
    fig, ax = plt.subplots(figsize=(10, 6))
    colors = ['blue', 'green', 'red', 'purple', 'orange', 'cyan']
    product_colors = {}
    product_labels = product_order.split(',')  # Split the product order string

    # 初始化每个产品的状态：当前工序索引、完成时间
    product_state = {
        prod: {
            "current_step": 0,
            "last_finish": 0
        } for prod in product_labels
    }
    for idx, product in enumerate(product_labels):
        product_colors[product] = colors[idx % len(colors)]

    # 机器的下一次可用时间
    machine_available = {machine: 0 for machine in processing_times}
    # 记录所有未完成的产品
    unfinished_products = set(product_labels)

    while unfinished_products:
        # 遍历所有未完成的产品，找出当前可以启动的工序
        scheduled = False
        for product in list(unfinished_products):
            state = product_state[product]
            step_idx = state["current_step"]
            product_machines = machine_assignment[product_labels.index(product)]
            
            # 如果当前产品还有未执行的工序
            if step_idx < len(product_machines):
                target_machine = product_machines[step_idx]
                # 工序的启动时间是产品前序工序完成时间和机器空闲时间的最大值
                start_time = max(state["last_finish"], machine_available[target_machine])
                process_time = processing_times[target_machine]
                
                # 绘制甘特图块
                ax.barh(target_machine, process_time, left=start_time, color=product_colors[product], edgecolor='black')
                
                # 更新状态
                state["last_finish"] = start_time + process_time
                machine_available[target_machine] = start_time + process_time
                state["current_step"] += 1
                
                # 如果当前产品所有工序完成，从未完成集合中移除
                if state["current_step"] >= len(product_machines):
                    unfinished_products.remove(product)
                
                scheduled = True
        
        # 防止死循环（理论上不会触发）
        if not scheduled:
            break

    # Add legend
    patches = [mpatches.Patch(color=color, label=product) for product, color in product_colors.items()]
    plt.legend(handles=patches)

    # Add labels and title
    ax.set_xlabel('时间')
    ax.set_ylabel('机器')
    ax.set_title('优化后的动态甘特图')
    plt.grid(True)
    plt.show()

修正说明 #

1. 状态跟踪：为每个产品维护当前待执行工序的索引和上一道工序的完成时间，确保产品的工序顺序不会被打乱。
2. 就绪工序调度：每次循环遍历所有未完成的产品，检查其当前工序对应的机器是否空闲，若满足条件则启动该工序，实现不同产品的工序并行。
3. 状态更新：每次启动工序后，更新产品的完成时间和机器的可用时间，当产品所有工序完成时，将其从未完成集合中移除，直到所有产品处理完毕。

内容的提问来源于stack exchange，提问作者Rafael Ferreira