【Python3】混合流水车间调度、遗传算法

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了【Python3】混合流水车间调度、遗传算法，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含8712字，纯文字阅读大概需要13分钟。

内容图文

1 准备
1.1 建立机器类

class Machine:
    def __init__(self, index, name=None):
        self.index = index
        self.name = name
        self.end = 0

    def clear(self):
        self.end = 0

1.2 建立加工任务类

class Task:
    def __init__(self, index, machine, duration, name=None):
        self.index = index
        self.machine = machine
        self.duration = duration
        self.name = name
        self.start = None
        self.end = None

    def clear(self):
        self.start = None
        self.end = None

1.3 建立工件类

class Job:
    def __init__(self, index, name=None):
        self.index = index
        self.name = name
        self.task = {}

    def clear(self):
        for i in self.task.keys():
            self.task[i].clear()

    @property
    def nop(self):
        return len(self.task)

    def add_task(self, machine, duration, name=None, index=None):
        if index is None:
            index = self.nop
        self.task[index] = Task(index, machine, duration, name)

    @property
    def start(self):
        return self.task[0].start

    @property
    def end(self):
        return self.task[self.nop-1].end

1.4 整合机器类、工件类及加工任务类

class Schedule(Code): # Code类为编码类，见2.1编码
    def __init__(self):
        self.job = {}
        self.machine = {}
        self.best_known = None
        self.time_unit = 1

    def clear(self):
        for i in self.job.keys():
            self.job[i].clear()
        for i in self.machine.keys():
            self.machine[i].clear()

    @property
    def n(self):
        return len(self.job)

    @property
    def m(self):
        return len(self.machine)

    @property
    def length(self):
        return sum([job.nop for job in self.job.values()])

    @property
    def makespan(self):
        return max([machine.end for machine in self.machine.values()])

    def add_machine(self, name=None, timetable=None, index=None):
        if index is None:
            index = self.m
        self.machine[index] = Machine(index, name, timetable)

    def add_job(self, due_date=None, name=None, index=None):
        if index is None:
            index = self.n
        self.job[index] = Job(index, due_date, name)

2 遗传算法

2.1 编码

class Code:
    @staticmethod
    def sequence_permutation(length):
        return np.random.permutation(length)

2.2 解码

class Hfsp(Schedule):
    def __init__(self):
        Schedule.__init__(self)

    def decode_permutation(self, code):
        self.clear()
        copy_code = deepcopy(code)
        mac, j = [[None for _ in range(job.nop)] for job in self.job.values()], 0
        while self.any_task_not_done():
            for i in copy_code:
                try:
                    a = self.job[i].task[j - 1].end
                except KeyError:
                    a = 0
                start, end, duration = [], [], []
                for k, p in zip(self.job[i].task[j].machine, self.job[i].task[j].duration):
                    early_start = max([a, self.machine[k].end])
                    start.append(early_start)
                    end.append(early_start + p)
                    duration.append(p)
                index_min_end = np.argwhere(np.array(end) == min(end))[:, 0]
                duration_in_min_end = np.array([duration[i] for i in index_min_end])
                choice_min_end_and_duration = np.argwhere(duration_in_min_end == np.min(duration_in_min_end))[:, 0]
                choice = index_min_end[np.random.choice(choice_min_end_and_duration, 1, replace=False)[0]]
                k, p = self.job[i].task[j].machine[choice], duration[choice]
                mac[i][j] = k
                self.job[i].task[j].start = start[choice]
                self.job[i].task[j].end = end[choice]
                if self.job[i].task[j].resumeable is not None:
                    res1, res2 = self.constrain_timetable(i, j, k, p, c)
                    if res1 is False:
                        continue
                    self.job[i].task[j].start = res1
                    self.job[i].task[j].end = res2
                if self.job[i].task[j].resumeable is not None:
                    self.constrain_timetable(i, j, k, p)
                if self.machine[k].end < self.job[i].task[j].end:
                    self.machine[k].end = self.job[i].task[j].end
                if self.job[i].task[0].limited_wait is not None:
                    while self.constrain_limited_wait(i, range(j, -1, -1), mac) is False:
                        pass
            copy_code = copy_code[np.argsort([self.job[i].task[j].start for i in copy_code])]
            j += 1
        return Info(code, self, mac) #Info类见2.3遗传操作

2.3  遗传操作

class Info:
    def __init__(self, code, schedule):
        self.code = code
        self.schedule = copy.deepcopy(schedule)
    
    def ga_crossover_sequence_pmx(self, info):
        code1 = deepcopy(self.code)
        code2 = deepcopy(info.code)
        a, b = np.random.choice(self.schedule.n, 2, replace=False)
        if a > b:
            a, b = b, a
        r_a_b = range(a, b)
        r_left = np.delete(range(self.schedule.n), r_a_b)
        left_1, left_2 = code2[r_left], code1[r_left]
        middle_1, middle_2 = code2[r_a_b], code1[r_a_b]
        code1[r_a_b], code2[r_a_b] = middle_2, middle_1
        mapping = [[], []]
        for i, j in zip(middle_1, middle_2):
            if j in middle_1 and i not in middle_2:
                index = np.argwhere(middle_1 == j)[0, 0]
                value = middle_2[index]
                while True:
                    if value in middle_1:
                        index = np.argwhere(middle_1 == value)[0, 0]
                        value = middle_2[index]
                    else:
                        break
                mapping[0].append(i)
                mapping[1].append(value)
            elif i in middle_2:
                pass
            else:
                mapping[0].append(i)
                mapping[1].append(j)
        for i, j in zip(mapping[0], mapping[1]):
            if i in left_1:
                left_1[np.argwhere(left_1 == i)[0, 0]] = j
            elif i in left_2:
                left_2[np.argwhere(left_2 == i)[0, 0]] = j
            if j in left_1:
                left_1[np.argwhere(left_1 == j)[0, 0]] = i
            elif j in left_2:
                left_2[np.argwhere(left_2 == j)[0, 0]] = i
        code1[r_left], code2[r_left] = left_1, left_2
        return code1, code2

    def ga_mutation_sequence_permutation_tpe(self):
        code = deepcopy(self.code)
        a = np.random.choice(range(self.schedule.n), 2, replace=False)
        code[a] = code[a[::-1]]
        return code

class Ga:
    def __init__(self, pop_size, rc, rm, max_generation, objective, schedule):
        self.pop_size = pop_size
        self.rc = rc
        self.rm = rm
        self.max_generation = max_generation
        self.objective = objective
        self.schedule = schedule
        self.best = [None, None, None]  # (info, objective, fitness)
        self.pop = [[], [], []]  # (info, objective, fitness)
        # (start, end, best_objective, best_fitness, worst_fitness, mean_fitness)
        self.record = [[], [], [], [], [], []]def clear(self):
        self.best = [None, None, None]
        self.pop = [[], [], []]
        self.record = [[], [], [], [], [], []]def update_individual(self, i, obj_new, info_new):
        fit_new = Utils.calculate_fitness(obj_new)
        if self.pop[1][i] >= obj_new or np.random.random() < 0.005:
            self.pop[0][i] = info_new
            self.pop[1][i] = obj_new
            self.pop[2][i] = fit_new
            for k in range(3):
                self.tabu_list[k][i] = []
            if self.best[1] >= obj_new:
                self.best[0] = info_new
                self.best[1] = obj_new
                self.best[2] = fit_new

    def init_best(self):
        self.best[2] = max(self.pop[2])
        index = self.pop[2].index(self.best[2])
        self.best[1] = self.pop[1][index]
        self.best[0] = deepcopy(self.pop[0][index])

    def show_generation(self, g):
        self.record[2].append(self.best[1])
        self.record[3].append(self.best[2])
        self.record[4].append(min(self.pop[2]))
        self.record[5].append(np.mean(self.pop[2]))
        Utils.print(
            "Generation {:<4} Runtime {:<8.4f} fBest: {:<.8f}, fWorst: {:<.8f}, fMean: {:<.8f}, gBest: {:<.2f} ".format(
                g, self.record[1][g] - self.record[0][g], self.record[3][g], self.record[4][g], self.record[5][g],
                self.record[2][g]))

    def reach_best_known_solution(self):
        if self.schedule.best_known is not None and self.best[1] <= self.schedule.best_known:
            return True
        return False

    def do_selection(self):
        a = np.array(self.pop[2]) / sum(self.pop[2])
        b = np.array([])
        for i in range(self.pop_size):
            b = np.append(b, sum(a[:i + 1]))
        pop = deepcopy(self.pop)
        tabu_list = deepcopy(self.tabu_list)
        for i in range(self.pop_size):
            j = np.argwhere(b > np.random.random())[0, 0]
            self.pop[0][i] = pop[0][j]
            self.pop[1][i] = pop[1][j]
            self.pop[2][i] = pop[2][j]
            for k in range(3):
                self.tabu_list[k][i] = tabu_list[k][j]
        c = self.pop[2].index(max(self.pop[2]))
        self.pop[0][c] = deepcopy(self.best[0])
        self.pop[1][c] = self.best[1]
        self.pop[2][c] = self.best[2]

    def do_init(self, pop=None):
        pass

    def do_crossover(self, i, j):
        pass

    def do_mutation(self, i):
        pass

    def do_tabu_search(self, i):
        pass

    def do_key_block_move(self, i):
        pass

    def do_evolution(self, pop=None):
        self.clear()
        self.do_init(pop)
        self.do_selection()
        for g in range(1, self.max_generation + 1):
            if self.reach_best_known_solution():
                break
            self.record[0].append(time.perf_counter())
            for i in range(self.pop_size):
                if self.reach_best_known_solution():
                    break
                if np.random.random() < self.rc:
                    j = np.random.choice(np.delete(np.arange(self.pop_size), i), 1, replace=False)[0]
                    self.do_crossover(i, j)
                if np.random.random() < self.rm:
                    self.do_mutation(i)
            self.do_selection()
            self.record[1].append(time.perf_counter())
            self.show_generation(g)

 

class GaFspHfsp(Ga):
    def __init__(self, pop_size, rc, rm, max_generation, objective, schedule):
        Ga.__init__(self, pop_size, rc, rm, max_generation, objective, schedule)

    def decode_update(self, i, code):
        info = self.schedule.decode_permutation(code)
        self.update_individual(i, self.objective(info), info)

    def do_init(self, pop=None):
        self.record[0].append(time.perf_counter())
        for i in range(self.pop_size):
            if pop is None:
                code = self.schedule.sequence_permutation(self.schedule.n)
            else:
                code = pop[0][i].code
            info = self.schedule.decode_permutation(code)
            self.pop[0].append(info)
            self.pop[1].append(self.objective(info))
            self.pop[2].append(Utils.calculate_fitness(self.pop[1][i]))
        self.init_best()
        self.record[1].append(time.perf_counter())
        self.show_generation(0)

    def do_crossover(self, i, j):
        code1, code2 = self.pop[0][i].ga_crossover_sequence_pmx(self.pop[0][j])
        self.decode_update(i, code1)
        self.decode_update(j, code2)

    def do_mutation(self, i):
        code1 = self.pop[0][i].ga_mutation_sequence_permutation_tpe()
        self.decode_update(i, code1)

 

 

 

 

内容总结

以上是互联网集市为您收集整理的【Python3】混合流水车间调度、遗传算法全部内容，希望文章能够帮你解决【Python3】混合流水车间调度、遗传算法所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

---