基于python对生产排程优化+柔性作业车间调度（源码）.zip

from data import * from multiprocess import * from pylab import * from copy import deepcopy import plotly.offline import plotly.express as px import matplotlib.pyplot as plt from itertools import permutations mpl.rcParams['font.sans-serif'] = ['SimHei'] class GA: """遗传算法类""" def __init__(self, data, strategy, population_size, crossover_rate, mutation_rate, select_rate, best_keep_rate, evolution_num, mutation_change_point): """ :param data: 传入遗传算法的对应数据类GAData :param strategy: 启发式策略，编号 :param population_size: 种群规模 :param crossover_rate: 交叉概率 :param mutation_rate: 变异概率 :param select_rate: 种群选择的比例 :param best_keep_rate: 保留最优个体的比例 :param evolution_num: 进化次数 :param mutation_change_point: 变异算子改变的时点 """ self.population_size = population_size # 种群规模 self.crossover_rate = crossover_rate # 交叉概率 self.mutation_rate = mutation_rate # 变异概率 self.select_rate = select_rate # 选择比例 self.best_keep_rate = best_keep_rate # 最优保留的数量 self.mutation_change_point = mutation_change_point self.evolution_num = evolution_num # 进化次数 self.cal = CalculateUtils() self.data = data self.job_id_list = data.job_id_list # 任务编号，list self.jobs = data.jobs # 任务集合 self.job_num = data.job_num # 任务总数 self.machine_id_list = data.machine_id_list self.machines = data.machines # 机器集合 self.machine_num = data.machine_num # 机器总数 self.chromosome_size = data.chromosome_size # 染色体长度 abs_path = os.path.abspath(__file__) _, filename = os.path.split(abs_path) filename = filename[:-3] self.file_name = filename + f"Strategy{strategy}PopulationSize{population_size}CR{crossover_rate}MR{mutation_rate}SR{select_rate}BR{best_keep_rate}MCP{mutation_change_point}EvolutionNum{evolution_num}" '''种群初始化''' def initialPopulation(self, chromosome_num): """编码，返回初始解array""" population = np.empty(shape=(chromosome_num, self.chromosome_size, 2)).astype(int) # 2表示一个是工序，一个是对应的机器 for i in range(chromosome_num): # 遍历种群中的所有染色体 # 染色体任务部分的数据生成 job_data = np.array(self.job_id_list) job_data = np.repeat(job_data, list(map(lambda x: x.procedure_num, self.jobs))) np.random.shuffle(job_data) # 染色体构造 for j in range(self.chromosome_size): # 遍历染色体上的所有基因，遍历job_data # 将任务部分的数据指定到染色体的相应基因位置 job_num = job_data[j] # 任务编号 population[i][j][0] = job_num # 随机选择各个任务的一个加工机器，并指定到染色体的相应基因位置 procedure_show_times = np.sum(job_data[0:j + 1] == job_num) # 某任务编号是第几次出现，从而得出这是该任务的第几道工序 population[i][j][1] = choice(list(filter(lambda x: x in self.machine_id_list, self.jobs.loc[job_num].machines[ procedure_show_times - 1]))) # 从该任务的工序对应的可选机器列表中随机选择一个 return population '''解码：①调度表，②目标值''' def decodeJobStrategy1(self, is_first_job_of_machine, machine_id, job_id): """对某机器上的任务进行解码""" # 数据准备 '''结果存储，生成dataframe''' start_time_list = [] end_time_list = [] machine_list = [] machine_status_list = [] procedure_list = [] category_id_list = [] order_id_list = [] product_id_list = [] num_list = [] '''结果存储，生成用于计算目标值的schedule_dataframe''' start_time_list_for_cal = [] end_time_list_for_cal = [] machine_list_for_cal = [] machine_status_list_for_cal = [] procedure_list_for_cal = [] category_id_list_for_cal = [] order_id_list_for_cal = [] product_id_list_for_cal = [] this_job = self.jobs.loc[job_id] this_machine = self.machines.loc[machine_id] order_batch = this_job.orders this_procedure = this_machine.procedure maintenance_day0 = this_machine.maintenance_day[0] # 机器当月的维修日期 maintenance_day1 = this_machine.maintenance_day[1] # 机器下月的维修日期 model_change_time = this_machine.model_change_time processing_time = this_machine.processing_time.loc[job_id] def linesAppend(status, start_time, end_time, category_ID, order_ID, product_ID, num_index): """往结果中增添一行""" if isinstance(start_time, list): num = len(start_time) else: num = 1 start_time = [start_time] end_time = [end_time] num_index = [num_index] status = [status] * num category_ID = [category_ID] * num order_ID = [order_ID] * num product_ID = [product_ID] * num machine = [machine_id] * num procedure = [this_procedure] * num start_time_list.extend(start_time) end_time_list.extend(end_time) machine_list.extend(machine) machine_status_list.extend(status) procedure_list.extend(procedure) category_id_list.extend(category_ID) order_id_list.extend(order_ID) product_id_list.extend(product_ID) num_list.extend(num_index) def linesAppendForCal(status, start_time, end_time, category_ID, order_ID, product_ID): """往结果中增添一行""" start_time_list_for_cal.append(start_time) end_time_list_for_cal.append(end_time) machine_list_for_cal.append(machine_id) machine_status_list_for_cal.append(status) procedure_list_for_cal.append(this_procedure) category_id_list_for_cal.append(category_ID) order_id_list_for_cal.append(order_ID) product_id_list_for_cal.append(product_ID) def saveSegment(start_time, end_time, status, this_order, product_ID): def getItemStartEndTime(index): item_start_time_ = start_time + index * processing_time item_end_time_ = start_time + (index + 1) * processing_time return item_start_time_, item_end_time_ if status == "生产": this_order_id = this_order.order_id item_index = list(range(this_order.product_number)) item_start_end_time = list(map(getItemStartEndTime, item_index)) item_start_time, item_end_time = zip(*item_start_end_time) item_start_time = list(item_start_time) item_end_time = list(item_end_time) linesAppend("生产", item_start_time, item_end_time, job_id, this_order_id, product_ID, item_index) linesAppendForCal('生产', start_time, end_time, job_id, this_order_id, product_ID) elif status == "维保": linesAppend('维保', start_time, end_time, '维保', '', '', '') linesAppendForCal('维保', start_time, end_time, '维保', '', '') elif status == "停机": linesAppend('停机', start_time, end_time, '停机', '', '', '