import pandas as pd import numpy as np from datetime import datetime, timedelta, time import os from typing import List, Dict, Optional # ====================== 基础参数 ====================== CHANGE_TIME = 3.5 # 换模时间（小时） REST_PERIODS = [ # 每日休息时段（小时） (12, 12.5), # 12:00-12:30 (16.5, 17), # 16:30-17:00 (0, 1) # 0:00-1:00 ] # ====================== 类定义 ====================== class Order: def __init__(self, order_id, product_id, quantity, delivery_date, capacity, weight, attr): self.order_id = order_id # 订单号 self.product_id = product_id # 产品号 self.quantity = quantity # 数量（pcs） self.delivery_date = delivery_date # 发货日期（datetime） self.capacity = capacity # 产能（pcs/h） self.weight = weight # 产品重量（g） self.processing_hours = quantity / capacity # 生产工时（h） self.start_time = None # 开始时间 self.end_time = None # 结束时间 self.machine = None # 分配机台 self.delay_days = 0 # 延期天数 self.attr = attr # 产品属性（1:易搭配，3:难生产） self.merged_id = None # 合单序号 self.plan_number = None # 计划序号 self.forward_plan = None # 前置任务计划序号 class MergedOrder: def __init__(self, product_id, total_quantity, delivery_dates, capacity, weight, attr, original_orders, merged_id): self.product_id = product_id # 产品号 self.total_quantity = total_quantity # 合并后总数量 self.delivery_dates = delivery_dates # 原始订单发货日期列表 self.capacity = capacity # 产能 self.weight = weight # 产品重量 self.processing_hours = total_quantity / capacity # 总生产工时 self.attr = attr # 产品属性 self.original_orders = original_orders # 原始订单对象列表 self.start_time = None # 开始时间 self.end_time = None # 结束时间 self.machine = None # 分配机台 self.merged_id = merged_id # 合单序号 self.plan_number = None # 计划序号 class Machine: def __init__(self, machine_id, initial_time=None): self.machine_id = machine_id # 机台号 self.available_time = initial_time or datetime(2025, 3, 1) # 初始可用时间 self.last_product = None # 上一生产产品 self.adjacent = [] # 相邻机台列表 self.schedule = [] # 排程计划（任务列表） self.last_plan_number = None # 上一任务计划序号 # ====================== 数据加载 ====================== def load_official_data(data_dir: str = "."): """ 加载附件数据 需修改data_dir为附件所在目录（例如："C:/Users/你的用户名/Desktop/附件"） """ try: # 读取附件1 attachment1 = pd.ExcelFile(os.path.join(data_dir, "附件1(1).xlsx")) orders_df = attachment1.parse( sheet_name="订单表", parse_dates=["发货日期(DeliveryDate)"], usecols=["订单号", "产品号", "订单数量", "发货日期(DeliveryDate)", "Weight/pcs(g)"] ) machine_initial_df = attachment1.parse( sheet_name="机台初始工作状态表", parse_dates=["生产开始时间"] ) holidays_df = attachment1.parse(sheet_name="放假日期表") holidays = {pd.to_datetime(row["放假日期"]).date() for _, row in holidays_df.iterrows()} # 读取附件2 attachment2 = pd.ExcelFile(os.path.join(data_dir, "附件2(1).xlsx")) product_capacity_df = attachment2.parse(sheet_name="产品工时计算参数表") product_machine_df = attachment2.parse(sheet_name="产品机台生产关系表") machine_relation_df = attachment2.parse(sheet_name="机台关系表") product_attr_df = attachment2.parse(sheet_name="产品属性表") # 构建数据映射 product_capacity = {row["产品号"]: row["Capacity（pcs/h）"] for _, row in product_capacity_df.iterrows()} product_weight = {row["产品号"]: row["Weight/pcs(g)"] for _, row in product_capacity_df.iterrows()} product_machines = {} for _, row in product_machine_df.iterrows(): valid_machines = [col for col in product_machine_df.columns if col != "产品号" and row[col] == 1] product_machines[row["产品号"]] = valid_machines machine_adjacent = {} for _, row in machine_relation_df.iterrows(): adjacent_machines = [col for col in machine_relation_df.columns if col != "机台号" and row[col] == 1] machine_adjacent[row["机台号"]] = adjacent_machines product_attrs = {row["产品号"]: row["属性"] for _, row in product_attr_df.iterrows()} # 补充订单表产能和重量 orders_df["Capacity（pcs/h）"] = orders_df["产品号"].map(product_capacity) orders_df["Weight/pcs(g)"] = orders_df["产品号"].map(product_weight) # 初始化机台 machine_initial_times = {row["机台号"]: row["生产开始时间"] for _, row in machine_initial_df.iterrows()} machines = [] for mid in machine_initial_times.keys(): machine = Machine(mid, machine_initial_times[mid]) machine.adjacent = machine_adjacent.get(mid, []) machines.append(machine) return orders_df, product_machines, product_attrs, machines, holidays except Exception as e: print(f"数据加载失败，请检查附件路径和格式：{e}") raise # ====================== 合单逻辑 ====================== def merge_orders(orders: List[Order], merge_days: int) -> tuple[List[MergedOrder], int]: merged_tasks = [] merge_count = 0 merged_id_counter = 1 product_groups = {} for order in orders: if order.product_id not in product_groups: product_groups[order.product_id] = [] product_groups[order.product_id].append(order) for product_id, group_orders in product_groups.items(): group_orders.sort(key=lambda x: x.delivery_date) current_merge = [group_orders[0]] for i in range(1, len(group_orders)): date_diff = (group_orders[i].delivery_date - current_merge[0].delivery_date).days if date_diff <= merge_days: current_merge.append(group_orders[i]) else: merged_tasks.append(MergedOrder( product_id=product_id, total_quantity=sum(o.quantity for o in current_merge), delivery_dates=[o.delivery_date for o in current_merge], capacity=current_merge[0].capacity, weight=current_merge[0].weight, attr=current_merge[0].attr, original_orders=current_merge, merged_id=merged_id_counter )) merge_count += 1 if len(current_merge) > 1 else 0 merged_id_counter += 1 current_merge = [group_orders[i]] merged_tasks.append(MergedOrder( product_id=product_id, total_quantity=sum(o.quantity for o in current_merge), delivery_dates=[o.delivery_date for o in current_merge], capacity=current_merge[0].capacity, weight=current_merge[0].weight, attr=current_merge[0].attr, original_orders=current_merge, merged_id=merged_id_counter )) if len(current_merge) > 1: merge_count += 1 merged_id_counter += 1 return merged_tasks, merge_count # ====================== 排程逻辑 ====================== def calculate_end_time(start_time: datetime, processing_hours: float, holidays: set) -> datetime: current = start_time remaining = processing_hours while remaining > 0: if current.date() in holidays: current = datetime.combine(current.date() + timedelta(days=1), time(0, 0)) continue day_start = datetime.combine(current.date(), time(0, 0)) intervals = [] prev_end = day_start for rest_start, rest_end in REST_PERIODS: rest_start_time = day_start + timedelta(hours=rest_start) rest_end_time = day_start + timedelta(hours=rest_end) if prev_end < rest_start_time: intervals.append((prev_end, rest_start_time)) prev_end = rest_end_time if prev_end < day_start + timedelta(hours=24): intervals.append((prev_end, day_start + timedelta(hours=24))) for (s, e) in intervals: if current < s: current = s if current >= e: continue available = (e - current).total_seconds() / 3600 use = min(remaining, available) current += timedelta(hours=use) remaining -= use if remaining <= 0: return current current = datetime.combine(current.date() + timedelta(days=1), time(0, 0)) return current def problem3_scheduling(merged_tasks: List[MergedOrder], product_machines: Dict, machines: List[Machine], holidays: set) -> tuple[List[Order], List[Dict]]: merged_tasks.sort(key=lambda x: (x.attr, min(x.delivery_dates))) machine_map = {m.machine_id: m for m in machines} plan_number = 1 detailed_results = [] for task in merged_tasks: candidate_machines = [m for m in machines if m.machine_id in product_machines.get(task.product_id, [])] if task.attr == 1: m03 = next((m for m in candidate_machines if m.machine_id == "M03"), None) if m03: candidate_machines = [m03] + [m for m in candidate_machines if m.machine_id != "M03"] best_machine = None best_end = None best_start = None for machine in candidate_machines: change = CHANGE_TIME if machine.last_product != task.product_id else 0 initial_start = machine.available_time + timedelta(hours=change) end_time = calculate_end_time(initial_start, task.processing_hours, holidays) adjusted_start = initial_start if task.attr == 3: for adj_id in machine.adjacent: adj_machine = machine_map.get(adj_id) if adj_machine and adj_machine.schedule: last_task = adj_machine.schedule[-1] if (last_task["product_attr"] == 3 and not (end_time <= last_task["start"] or adjusted_start >= last_task["end"])): adjusted_start = max(adjusted_start, last_task["end"]) end_time = calculate_end_time(adjusted_start, task.processing_hours, holidays) if best_end is None or end_time < best_end: best_end = end_time best_start = adjusted_start best_machine = machine if best_machine: task.start_time = best_start task.end_time = best_end task.machine = best_machine.machine_id task.plan_number = plan_number for original_order in task.original_orders: original_order.start_time = best_start original_order.end_time = best_end original_order.machine = best_machine.machine_id original_order.merged_id = task.merged_id original_order.plan_number = plan_number original_order.forward_plan = best_machine.last_plan_number original_order.delay_days = max(0, (best_end.date() - original_order.delivery_date.date()).days) detailed_results.append({ "计划序号（Plan Number）": plan_number, "生产计划安排机台号": best_machine.machine_id, "订单号PO": original_order.order_id, "产品号": original_order.product_id, "需要工时 (Requested Time--Hours)": round(original_order.processing_hours, 2), "生产计划开始时间": original_order.start_time.strftime("%Y-%m-%d %H:%M:%S"), "生产计划预计完成时间": original_order.end_time.strftime("%Y-%m-%d %H:%M:%S"), "发货日期(DeliveryDate)": original_order.delivery_date.strftime("%Y-%m-%d"), "订单数量": original_order.quantity, "Weight/pcs(g)": original_order.weight, "Capacity（pcs/h）": original_order.capacity, "最迟开始时间 (Late Start Time)": (original_order.delivery_date - timedelta( hours=original_order.processing_hours )).strftime("%Y-%m-%d %H:%M:%S"), "合单序号（Joined Order Number）": task.merged_id, "前置任务生产计划序号": original_order.forward_plan or "", "是否延期": "是" if original_order.delay_days > 0 else "否", "延期天数": original_order.delay_days }) best_machine.available_time = best_end best_machine.last_product = task.product_id best_machine.last_plan_number = plan_number best_machine.schedule.append({ "product": task.product_id, "start": best_start, "end": best_end, "product_attr": task.attr }) plan_number += 1 return detailed_results # ====================== 导出结果 ====================== def export_results(detailed_results: List[Dict], output_path: str = "附件5_生产排程计划表.xlsx"): """ 导出完整排程结果到Excel 需修改output_path为期望的导出路径（例如："C:/Users/你的用户名/Desktop/附件5_结果.xlsx"） """ df = pd.DataFrame(detailed_results) column_order = [ "计划序号（Plan Number）", "生产计划安排机台号", "订单号PO", "产品号", "需要工时 (Requested Time--Hours)", "生产计划开始时间", "生产计划预计完成时间", "发货日期(DeliveryDate)", "订单数量", "Weight/pcs(g)", "Capacity（pcs/h）", "最迟开始时间 (Late Start Time)", "合单序号（Joined Order Number）", "前置任务生产计划序号", "是否延期", "延期天数" ] df = df[column_order] df.to_excel(output_path, index=False) print(f"结果已导出至：{os.path.abspath(output_path)}") # ====================== 主函数 ====================== def main(): # ---------------------- 需要您修改的部分 ---------------------- # 1. 附件所在目录：将"."改为附件实际存放路径（例如："C:/Users/你的用户名/Desktop/附件"） DATA_DIR = "." # 2. 导出文件路径：将"附件5_生产排程计划表.xlsx"改为期望的导出路径（例如："C:/Users/你的用户名/Desktop/结果.xlsx"） OUTPUT_PATH = "附件5_生产排程计划表.xlsx" # 3. 合单天数：可根据需要修改（当前为7天，可改为15或30天） MERGE_DAYS = 7 # ------------------------------------------------------------- try: orders_df, product_machines, product_attrs, machines, holidays = load_official_data(DATA_DIR) original_orders = [] for _, row in orders_df.iterrows(): try: order = Order( order_id=row["订单号"], product_id=row["产品号"], quantity=row["订单数量"], delivery_date=row["发货日期(DeliveryDate)"], capacity=row["Capacity（pcs/h）"], weight=row["Weight/pcs(g)"], attr=product_attrs.get(row["产品号"], 2) ) original_orders.append(order) except Exception as e: print(f"订单初始化失败: {e} - 行数据: {row}") if not original_orders: raise ValueError("没有有效订单，请检查数据格式") merged_tasks, merge_count = merge_orders(original_orders, MERGE_DAYS) print(f"合单完成，共合并 {merge_count} 次") detailed_results = problem3_scheduling(merged_tasks, product_machines, machines, holidays) print(f"排程完成，共生成 {len(detailed_results)} 条计划") export_results(detailed_results, OUTPUT_PATH) except Exception as e: print(f"程序执行失败: {e}") if __name__ == "__main__": main() 最终结果呈现应该要有计划序号生产计划安排机台号，订单号，产品号需要工时生产，计划开始时间，生产计划预计完成时间，发货日期，订单数量，重量，产能，最迟开始时间，合单序号，前置任务生产计划序号，是否延期，延期天数

<think>我们有一个已经编写好的程序，用于解决生产排程问题。现在，我们需要按照要求导出结果，并且确保结果中包含指定的列。 根据要求，最终结果需要包含以下列： 1. 计划序号（Plan Number） 2. 生产计划安排机台号 3. 订单号PO 4. 产品号 5. 需要工时 (Requested Time--Hours) 6. 生产计划开始时间 7. 生产计划预计完成时间 8. 发货日期(DeliveryDate) 9. 订单数量 10. Weight/pcs(g) 11. Capacity（pcs/h） 12. 最迟开始时间 (Late Start Time) 13. 合单序号（Joined Order Number） 14. 前置任务生产计划序号 15. 是否延期 16. 延期天数 在程序中的`problem3_scheduling`函数里，我们已经为每个原始订单生成了一个字典，包含了这些信息，并添加到`detailed_results`列表中。 然后通过`export_results`函数导出为Excel。 但是，我们注意到在`detailed_results`的字典中，我们使用了以下键名： - "计划序号（Plan Number）": plan_number - "生产计划安排机台号": best_machine.machine_id - "订单号PO": original_order.order_id - "产品号": original_order.product_id - "需要工时 (Requested Time--Hours)": round(original_order.processing_hours, 2) - "生产计划开始时间": original_order.start_time.strftime("%Y-%m-%d %H:%M:%S") - "生产计划预计完成时间": original_order.end_time.strftime("%Y-%m-%d %H:%M:%S") - "发货日期(DeliveryDate)": original_order.delivery_date.strftime("%Y-%m-%d") - "订单数量": original_order.quantity - "Weight/pcs(g)": original_order.weight - "Capacity（pcs/h）": original_order.capacity - "最迟开始时间 (Late Start Time)": (original_order.delivery_date - timedelta(hours=original_order.processing_hours)).strftime("%Y-%m-%d %H:%M:%S") - "合单序号（Joined Order Number）": task.merged_id - "前置任务生产计划序号": original_order.forward_plan or "" (如果为None则用空字符串) - "是否延期": "是" if original_order.delay_days > 0 else "否" - "延期天数": original_order.delay_days 因此，我们只需要确保在导出时，列的顺序按照上述要求排列即可。 在`export_results`函数中，我们已经指定了列的顺序，所以导出没有问题。 但是，我们注意到在`main`函数中，我们调用了`export_results(detailed_results, OUTPUT_PATH)`，其中`detailed_results`就是包含所有字典的列表。 因此，我们只需要运行程序，并确保附件路径正确，即可生成结果。 另外，我们注意到在`load_official_data`函数中，我们读取了附件1和附件2，并进行了数据整合。在运行前，请确保将`DATA_DIR`设置为附件所在目录。 现在，我们提供一个完整的代码，并提示用户修改数据目录和输出路径。 注意：由于问题3中要求合单天数为7天，我们在主函数中设置了`MERGE_DAYS = 7`，但也可以根据需要修改为15或30天。 运行程序后，将在指定路径生成Excel文件。 如果遇到任何错误，程序会捕获异常并打印错误信息。 由于代码较长，我们不再重复贴出整个代码，但请注意以下几点： 1. 在`load_official_data`中，我们使用`parse_dates`参数解析日期列。 2. 在合并订单时，我们按照产品号分组，并在每个产品内按发货日期排序，然后合并相邻7天内的订单。 3. 排程时，我们考虑了换模时间、休息时段、节假日、相邻机台不能同时生产属性3的产品等约束。 4. 在计算结束时间时，我们使用了`calculate_end_time`函数，该函数会跳过休息时间和节假日。 现在，我们假设用户已经将代码保存为Python文件，并准备好附件数据，然后运行即可。 如果用户需要修改合单天数，可以在主函数中修改`MERGE_DAYS`变量。 最后，我们提供一次完整的代码，但为了简洁，我们只给出主函数和必要的部分，因为整个代码已经很长了。但请注意，下面的代码是完整的，只是省略了类定义和函数定义（但实际上在之前的代码中已经给出）。 但是，为了确保清晰，我们将整个代码结构再次完整写出，但省略了类定义和函数内部的注释（因为前面已经详细注释）。 由于代码过长，我们建议用户直接使用之前提供的完整代码。 注意：在运行之前，请确保安装了pandas和openpyxl（用于写入Excel）。 安装命令：pip install pandas openpyxl 如果遇到路径问题，请使用绝对路径。 下面，我们再次提供完整代码，但为了减少重复，我们只给出关键部分，并确保代码可以运行。 由于我们之前已经提供了完整代码，这里不再重复，但为了确保完整性，我们将关键部分重新整理如下：</think>我将为您提供一个完整的生产排程解决方案，包含详细的代码实现和解释。该方案基于您提供的需求，实现了订单合并、机台调度和生产计划生成功能。 ```python import pandas as pd import numpy as np from datetime import datetime, timedelta, time import os from typing import List, Dict, Optional # ====================== 基础参数配置 ====================== CHANGE_TIME = 3.5 # 换模时间（小时） REST_PERIODS = [ # 每日休息时段（小时） (12, 12.5), # 12:00-12:30 (16.5, 17), # 16:30-17:00 (0, 1) # 0:00-1:00 ] # ====================== 核心类定义 ====================== class Order: """订单类，表示单个生产订单""" def __init__(self, order_id, product_id, quantity, delivery_date, capacity, weight, attr): self.order_id = order_id # 订单号 self.product_id = product_id # 产品号 self.quantity = quantity # 数量（pcs） self.delivery_date = delivery_date # 发货日期（datetime） self.capacity = capacity # 产能（pcs/h） self.weight = weight # 产品重量（g） # 计算生产所需工时 self.processing_hours = quantity / capacity if capacity > 0 else 0 self.start_time = None # 生产开始时间 self.end_time = None # 生产结束时间 self.machine = None # 分配机台 self.delay_days = 0 # 延期天数 self.attr = attr # 产品属性（1:易搭配，3:难生产） self.merged_id = None # 合单序号 self.plan_number = None # 计划序号 self.forward_plan = None # 前置任务计划序号 class MergedOrder: """合并订单类，表示合并后的生产任务""" def __init__(self, product_id, total_quantity, delivery_dates, capacity, weight, attr, original_orders, merged_id): self.product_id = product_id # 产品号 self.total_quantity = total_quantity # 合并后总数量 self.delivery_dates = delivery_dates # 原始订单发货日期列表 self.capacity = capacity # 产能 self.weight = weight # 产品重量 # 计算总生产工时 self.processing_hours = total_quantity / capacity if capacity > 0 else 0 self.attr = attr # 产品属性 self.original_orders = original_orders # 原始订单对象列表 self.start_time = None # 开始时间 self.end_time = None # 结束时间 self.machine = None # 分配机台 self.merged_id = merged_id # 合单序号 self.plan_number = None # 计划序号 class Machine: """机台类，表示生产设备及其状态""" def __init__(self, machine_id, initial_time=None): self.machine_id = machine_id # 机台号 # 初始可用时间，默认为2025年3月1日 self.available_time = initial_time or datetime(2025, 3, 1) self.last_product = None # 上一生产产品 self.adjacent = [] # 相邻机台列表 self.schedule = [] # 排程计划（任务列表） self.last_plan_number = None # 上一任务计划序号 # ====================== 数据加载函数 ====================== def load_official_data(data_dir: str = "."): """ 加载附件数据 :param data_dir: 附件所在目录 :return: 订单数据、产品-机台关系、产品属性、机台列表、节假日集合 """ try: # 1. 读取附件1 - 订单和机台信息 attachment1 = pd.ExcelFile(os.path.join(data_dir, "附件1(1).xlsx")) # 加载订单表 orders_df = attachment1.parse( sheet_name="订单表", parse_dates=["发货日期(DeliveryDate)"], usecols=["订单号", "产品号", "订单数量", "发货日期(DeliveryDate)", "Weight/pcs(g)"] ) # 加载机台初始状态表 machine_initial_df = attachment1.parse( sheet_name="机台初始工作状态表", parse_dates=["生产开始时间"] ) # 加载放假日期表 holidays_df = attachment1.parse(sheet_name="放假日期表") holidays = {pd.to_datetime(row["放假日期"]).date() for _, row in holidays_df.iterrows()} # 2. 读取附件2 - 产品参数和关系 attachment2 = pd.ExcelFile(os.path.join(data_dir, "附件2(1).xlsx")) # 加载产品工时参数 product_capacity_df = attachment2.parse(sheet_name="产品工时计算参数表") # 加载产品-机台关系 product_machine_df = attachment2.parse(sheet_name="产品机台生产关系表") # 加载机台关系 machine_relation_df = attachment2.parse(sheet_name="机台关系表") # 加载产品属性 product_attr_df = attachment2.parse(sheet_name="产品属性表") # 3. 构建数据映射关系 # 产品产能映射 product_capacity = {row["产品号"]: row["Capacity（pcs/h）"] for _, row in product_capacity_df.iterrows()} # 产品重量映射 product_weight = {row["产品号"]: row["Weight/pcs(g)"] for _, row in product_capacity_df.iterrows()} # 产品可生产机台映射 product_machines = {} for _, row in product_machine_df.iterrows(): valid_machines = [col for col in product_machine_df.columns if col != "产品号" and row[col] == 1] product_machines[row["产品号"]] = valid_machines # 机台相邻关系映射 machine_adjacent = {} for _, row in machine_relation_df.iterrows(): adjacent_machines = [col for col in machine_relation_df.columns if col != "机台号" and row[col] == 1] machine_adjacent[row["机台号"]] = adjacent_machines # 产品属性映射 product_attrs = {row["产品号"]: row["属性"] for _, row in product_attr_df.iterrows()} # 4. 补充订单表数据 orders_df["Capacity（pcs/h）"] = orders_df["产品号"].map(product_capacity) orders_df["Weight/pcs(g)"] = orders_df["产品号"].map(product_weight) # 5. 初始化机台对象 machine_initial_times = {row["机台号"]: row["生产开始时间"] for _, row in machine_initial_df.iterrows()} machines = [] for mid in machine_initial_times.keys(): machine = Machine(mid, machine_initial_times[mid]) machine.adjacent = machine_adjacent.get(mid, []) machines.append(machine) return orders_df, product_machines, product_attrs, machines, holidays except Exception as e: print(f"数据加载失败，请检查附件路径和格式：{e}") raise # ====================== 订单合并逻辑 ====================== def merge_orders(orders: List[Order], merge_days: int) -> tuple[List[MergedOrder], int]: """ 合并相同产品的订单 :param orders: 订单列表 :param merge_days: 合并天数阈值 :return: 合并任务列表, 合并次数 """ merged_tasks = [] merge_count = 0 merged_id_counter = 1 # 按产品分组 product_groups = {} for order in orders: if order.product_id not in product_groups: product_groups[order.product_id] = [] product_groups[order.product_id].append(order) # 对每个产品组的订单按发货日期排序 for product_id, group_orders in product_groups.items(): group_orders.sort(key=lambda x: x.delivery_date) current_merge = [group_orders[0]] # 遍历该产品组的所有订单 for i in range(1, len(group_orders)): # 计算日期差 date_diff = (group_orders[i].delivery_date - current_merge[0].delivery_date).days if date_diff <= merge_days: current_merge.append(group_orders[i]) else: # 创建合并任务 merged_tasks.append(create_merged_task( product_id, current_merge, merged_id_counter )) # 更新合并计数 merge_count += 1 if len(current_merge) > 1 else 0 merged_id_counter += 1 current_merge = [group_orders[i]] # 处理最后一组订单 merged_tasks.append(create_merged_task( product_id, current_merge, merged_id_counter )) if len(current_merge) > 1: merge_count += 1 merged_id_counter += 1 return merged_tasks, merge_count def create_merged_task(product_id, orders, merged_id): """创建合并任务对象""" return MergedOrder( product_id=product_id, total_quantity=sum(o.quantity for o in orders), delivery_dates=[o.delivery_date for o in orders], capacity=orders[0].capacity, weight=orders[0].weight, attr=orders[0].attr, original_orders=orders, merged_id=merged_id ) # ====================== 生产调度核心逻辑 ====================== def calculate_end_time(start_time: datetime, processing_hours: float, holidays: set) -> datetime: """ 计算考虑休息时间和节假日的实际结束时间 :param start_time: 开始时间 :param processing_hours: 需要工时 :param holidays: 节假日集合 :return: 实际结束时间 """ current = start_time remaining = processing_hours while remaining > 0: # 检查是否为节假日 if current.date() in holidays: current = datetime.combine(current.date() + timedelta(days=1), time(0, 0)) continue # 计算当天的有效工作时间段 day_start = datetime.combine(current.date(), time(0, 0)) intervals = [] prev_end = day_start # 根据休息时段划分工作时间段 for rest_start, rest_end in REST_PERIODS: rest_start_time = day_start + timedelta(hours=rest_start) rest_end_time = day_start + timedelta(hours=rest_end) if prev_end < rest_start_time: intervals.append((prev_end, rest_start_time)) prev_end = rest_end_time if prev_end < day_start + timedelta(hours=24): intervals.append((prev_end, day_start + timedelta(hours=24))) # 在每个时间段内工作 for (s, e) in intervals: if current < s: current = s if current >= e: continue # 计算该时间段内可用时间 available = (e - current).total_seconds() / 3600 use = min(remaining, available) current += timedelta(hours=use) remaining -= use if remaining <= 0: return current # 转到下一天 current = datetime.combine(current.date() + timedelta(days=1), time(0, 0)) return current def problem3_scheduling(merged_tasks: List[MergedOrder], product_machines: Dict, machines: List[Machine], holidays: set) -> tuple[List[Order], List[Dict]]: """ 生产调度主函数 :param merged_tasks: 合并任务列表 :param product_machines: 产品-机台关系 :param machines: 机台列表 :param holidays: 节假日集合 :return: 详细排程结果 """ # 任务排序：属性优先（易搭配在前），然后按最早发货日期 merged_tasks.sort(key=lambda x: (x.attr, min(x.delivery_dates))) machine_map = {m.machine_id: m for m in machines} plan_number = 1 detailed_results = [] for task in merged_tasks: # 获取可生产该产品的机台 candidate_machines = [m for m in machines if m.machine_id in product_machines.get(task.product_id, [])] # 特殊处理：易搭配产品优先分配到M03机台 if task.attr == 1: m03 = next((m for m in candidate_machines if m.machine_id == "M03"), None) if m03: candidate_machines = [m03] + [m for m in candidate_machines if m.machine_id != "M03"] best_machine = None best_end = None best_start = None # 评估每个候选机台 for machine in candidate_machines: # 计算换模时间 change = CHANGE_TIME if machine.last_product != task.product_id else 0 initial_start = machine.available_time + timedelta(hours=change) # 计算实际结束时间 end_time = calculate_end_time(initial_start, task.processing_hours, holidays) adjusted_start = initial_start # 特殊处理：难生产产品需考虑相邻机台 if task.attr == 3: for adj_id in machine.adjacent: adj_machine = machine_map.get(adj_id) if adj_machine and adj_machine.schedule: last_task = adj_machine.schedule[-1] # 检查时间冲突 if (last_task["product_attr"] == 3 and not (end_time <= last_task["start"] or adjusted_start >= last_task["end"])): # 调整开始时间以避免冲突 adjusted_start = max(adjusted_start, last_task["end"]) end_time = calculate_end_time(adjusted_start, task.processing_hours, holidays) # 选择最早结束的机台 if best_end is None or end_time < best_end: best_end = end_time best_start = adjusted_start best_machine = machine # 分配任务到最佳机台 if best_machine: # 更新任务信息 task.start_time = best_start task.end_time = best_end task.machine = best_machine.machine_id task.plan_number = plan_number # 更新原始订单信息 for original_order in task.original_orders: original_order.start_time = best_start original_order.end_time = best_end original_order.machine = best_machine.machine_id original_order.merged_id = task.merged_id original_order.plan_number = plan_number original_order.forward_plan = best_machine.last_plan_number # 计算延期天数 original_order.delay_days = max(0, (best_end.date() - original_order.delivery_date.date()).days) # 构建结果字典 detailed_results.append({ "计划序号（Plan Number）": plan_number, "生产计划安排机台号": best_machine.machine_id, "订单号PO": original_order.order_id, "产品号": original_order.product_id, "需要工时 (Requested Time--Hours)": round(original_order.processing_hours, 2), "生产计划开始时间": original_order.start_time.strftime("%Y-%m-%d %H:%M:%S"), "生产计划预计完成时间": original_order.end_time.strftime("%Y-%m-%d %H:%M:%S"), "发货日期(DeliveryDate)": original_order.delivery_date.strftime("%Y-%m-%d"), "订单数量": original_order.quantity, "Weight/pcs(g)": original_order.weight, "Capacity（pcs/h）": original_order.capacity, "最迟开始时间 (Late Start Time)": (original_order.delivery_date - timedelta( hours=original_order.processing_hours )).strftime("%Y-%m-%d %H:%M:%S"), "合单序号（Joined Order Number）": task.merged_id, "前置任务生产计划序号": original_order.forward_plan or "", "是否延期": "是" if original_order.delay_days > 0 else "否", "延期天数": original_order.delay_days }) # 更新机台状态 best_machine.available_time = best_end best_machine.last_product = task.product_id best_machine.last_plan_number = plan_number best_machine.schedule.append({ "product": task.product_id, "start": best_start, "end": best_end, "product_attr": task.attr }) plan_number += 1 return detailed_results # ====================== 结果导出函数 ====================== def export_results(detailed_results: List[Dict], output_path: str = "附件5_生产排程计划表.xlsx"): """ 导出完整排程结果到Excel :param detailed_results: 详细排程结果 :param output_path: 输出文件路径 """ # 创建DataFrame df = pd.DataFrame(detailed_results) # 定义列顺序 column_order = [ "计划序号（Plan Number）", "生产计划安排机台号", "订单号PO", "产品号", "需要工时 (Requested Time--Hours)", "生产计划开始时间", "生产计划预计完成时间", "发货日期(DeliveryDate)", "订单数量", "Weight/pcs(g)", "Capacity（pcs/h）", "最迟开始时间 (Late Start Time)", "合单序号（Joined Order Number）", "前置任务生产计划序号", "是否延期", "延期天数" ] # 按指定顺序排列列 df = df[column_order] # 导出到Excel df.to_excel(output_path, index=False) print(f"结果已导出至：{os.path.abspath(output_path)}") # ====================== 主执行函数 ====================== def main(): """主函数，执行整个排程流程""" # ---------------------- 用户配置区域 ---------------------- # 1. 附件所在目录：将"."改为附件实际存放路径 DATA_DIR = "." # 2. 导出文件路径：设置结果输出路径 OUTPUT_PATH = "附件5_生产排程计划表.xlsx" # 3. 合单天数：可根据需要修改（7/15/30天） MERGE_DAYS = 7 # -------------------------------------------------------- try: print("开始加载数据...") # 加载数据 orders_df, product_machines, product_attrs, machines, holidays = load_official_data(DATA_DIR) print("初始化订单对象...") # 初始化订单对象 original_orders = [] for _, row in orders_df.iterrows(): order = Order( order_id=row["订单号"], product_id=row["产品号"], quantity=row["订单数量"], delivery_date=row["发货日期(DeliveryDate)"], capacity=row["Capacity（pcs/h）"], weight=row["Weight/pcs(g)"], attr=product_attrs.get(row["产品号"], 2) # 默认为普通产品 ) original_orders.append(order) if not original_orders: raise ValueError("没有有效订单，请检查数据格式") print(f"开始订单合并（合单天数：{MERGE_DAYS}天）...") # 合并订单 merged_tasks, merge_count = merge_orders(original_orders, MERGE_DAYS) print(f"合单完成，共合并 {merge_count} 次，生成 {len(merged_tasks)} 个合并任务") print("开始生产排程...") # 执行排程 detailed_results = problem3_scheduling(merged_tasks, product_machines, machines, holidays) print(f"排程完成，共生成 {len(detailed_results)} 条生产计划") print("导出结果...") # 导出结果 export_results(detailed_results, OUTPUT_PATH) print("程序执行完成！") except Exception as e: print(f"程序执行失败: {e}") import traceback traceback.print_exc() if __name__ == "__main__": main() ``` ### 代码详细解释 #### 1. 基础参数配置 - `CHANGE_TIME`: 换模时间（3.5小时） - `REST_PERIODS`: 每日休息时段（中午12:00-12:30，下午16:30-17:00，凌晨0:00-1:00） #### 2. 核心类定义 - **Order类**：表示单个生产订单，包含订单号、产品号、数量、发货日期等关键信息 - **MergedOrder类**：表示合并后的生产任务，包含原始订单列表和合并信息 - **Machine类**：表示生产机台，包含机台状态、排程计划和可用时间 #### 3. 数据加载函数 (`load_official_data`) - 从Excel附件加载订单、机台、产品和假期数据 - 构建产品-产能、产品-机台、机台-相邻关系等映射 - 返回结构化数据供后续处理 #### 4. 订单合并逻辑 (`merge_orders`) - 按产品分组，相同产品订单按发货日期排序 - 合并发货日期在指定天数内的订单（默认7天） - 创建合并任务对象，保留原始订单引用 #### 5. 生产调度核心逻辑 (`problem3_scheduling`) - **任务排序**：优先处理易搭配产品（属性1），然后按最早发货日期 - **机台选择**： - 易搭配产品优先分配到M03机台 - 考虑换模时间（不同产品切换需3.5小时） - 难生产产品（属性3）避免相邻机台同时生产 - **时间计算**：使用`calculate_end_time`考虑休息时间和节假日 - **结果生成**：为每个订单创建详细排程记录 #### 6. 时间计算函数 (`calculate_end_time`) - 考虑每日休息时段 - 跳过节假日 - 精确计算实际生产结束时间 #### 7. 结果导出函数 (`export_results`) - 生成包含16个指定字段的DataFrame - 按指定列顺序导出为Excel文件 #### 8. 主执行流程 (`main`) 1. 加载数据 2. 初始化订单对象 3. 合并订单（默认7天） 4. 执行生产排程 5. 导出结果到Excel ### 使用说明 1. 将附件1和附件2放在指定目录（默认为当前目录） 2. 修改`DATA_DIR`变量为实际附件路径 3. 设置`OUTPUT_PATH`为期望的输出文件路径 4. 可选：调整`MERGE_DAYS`（7/15/30天） 5. 运行程序，生成排程结果 ### 输出结果说明 结果Excel文件包含以下16个字段： 1. 计划序号（Plan Number） 2. 生产计划安排机台号 3. 订单号PO 4. 产品号 5. 需要工时 (Requested Time--Hours) 6. 生产计划开始时间 7. 生产计划预计完成时间 8. 发货日期(DeliveryDate) 9. 订单数量 10. Weight/pcs(g) 11. Capacity（pcs/h） 12. 最迟开始时间 (Late Start Time) 13. 合单序号（Joined Order Number） 14. 前置任务生产计划序号 15. 是否延期 16. 延期天数