【Python3】作业车间调度，SPT规则，LPT规则

程序员文章站 2022-07-09 23:22:29
作业车间调度，SPT规则，LPT规则...
部分程序

import copy

import numpy as np
import unittest
from jsp import Jsp, Utils

deepcopy = copy.deepcopy


class Operation:
    """
    Operation class
    """

    def __init__(self, job, operation, machine, duration, ):
        """

        :param job:
        :param operation:
        :param machine:
        :param duration:
        """
        self.job = job
        self.operation = operation
        self.machine = machine
        self.duration = duration

    def __str__(self, ):
        return "{%s,%s,%s,%s}" % (
            self.job,
            self.operation,
            self.machine,
            self.duration,
        )


class DataProduce:
    """
    All task (job, operation)
    """

    def __init__(self, ):
        self.task_dict = {}

    def add_task(self, i, j, k, p, ):
        """

        :param i: job
        :param j: operation
        :param k: machine
        :param p: duration
        :return:
        """
        self.task_dict[(i, j)] = Operation(i, j, k, p)

    def __str__(self, ):
        string = {}
        for key, task in self.task_dict.items():
            string[key] = str(task)
        return str(string)


class JSPRule:
    """
    Job shop scheduling problem
    Scheduling rules
    """

    def __init__(self, n, m, data, ):
        """

        :param n: number of job
        :param m: number of machine
        :param data: DataProduce
        """
        self.n = n
        self.m = m
        self.data = data
        self.task_dict = {}
        self.task_machine = []
        self.task_duration = []
        self.task_wait = []
        self.task_done = []
        self.task_operation = []

    def update(self, choice, ):
        """

        :param choice:
        :return:
        """
        decision = self.task_wait[choice]
        # update task_dict
        self.task_dict.pop(decision)
        # update task_wait，task_machine，task_duration，task_operation
        self.task_wait.pop(choice)
        self.task_machine.pop(choice)
        self.task_duration.pop(choice)
        # update task_done
        job = decision[0]
        self.task_done.append(job)
        # update task_operation，task_wait
        self.task_operation[job] += 1
        if self.task_operation[job] < self.m:
            self.task_wait.append((job, self.task_operation[job]))

    def reset(self, ):
        self.task_dict = deepcopy(self.data.task_dict)
        self.task_done = []
        self.task_wait = [(i, 0) for i in range(self.n)]
        self.task_operation = [0 for _ in range(self.n)]

    @property
    def spt(self, ):
        """

        :return:
        """
        self.reset()
        while self.task_dict.__len__() > 0:
            self.task_machine = [self.task_dict[task].machine for task in self.task_wait]
            self.task_duration = [self.task_dict[task].duration for task in self.task_wait]
            machines = set(self.task_machine)
            eq_machines = []
            for k in machines:
                eq_machines.append(self.task_machine.count(k))
            for i, j in enumerate(machines):
                if eq_machines[i] == 1:
                    self.update(self.task_machine.index(j))
                else:
                    index = np.argwhere(np.array(self.task_machine) == j)[:, 0]
                    proc = np.array(self.task_duration)[index]
                    choice = np.argmin(proc)
                    ext_choice = np.argwhere(proc == proc[choice])[:, 0]
                    for a, b in enumerate(ext_choice):
                        self.update(index[b] - a)
        return self.task_done

    @property
    def lpt(self, ):
        """

        :return:
        """
        self.reset()
        while self.task_dict.__len__() > 0:
            self.task_machine = [self.task_dict[task].machine for task in self.task_wait]
            self.task_duration = [self.task_dict[task].duration for task in self.task_wait]
            machines = set(self.task_machine)
            eq_machines = []
            for k in machines:
                eq_machines.append(self.task_machine.count(k))
            for i, j in enumerate(machines):
                if eq_machines[i] == 1:
                    self.update(self.task_machine.index(j))
                else:
                    index = np.argwhere(np.array(self.task_machine) == j)[:, 0]
                    proc = np.array(self.task_duration)[index]
                    choice = np.argmax(proc)
                    ext_choice = np.argwhere(proc == proc[choice])[:, 0]
                    for a, b in enumerate(ext_choice):
                        self.update(index[b] - a)
        return self.task_done


class TestFunc(unittest.TestCase):
    def test_spt(self, ):
        ops = np.array([
            [0, 2, 1],
            [2, 1, 0],
            [0, 2, 1]
        ])
        prt = np.array([
            [1, 2, 3],
            [2, 2, 4],
            [3, 1, 2]
        ])
        n, m = ops.shape[0], ops.shape[1]
        # 预期结果
        # spt = [0, 1, 2, 1, 0, 1, 0, 2, 2]
        # lpt = [2, 1, 0, 1, 2, 1, 2, 0, 0]
        # 随机测试
        # n, m, low, high = 4, 3, 1, 6
        # ops, prt = Utils.crt_data_jsp(n, m, low, high)
        # print(ops, "# ops")
        # print(prt, "# prt")
        task = DataProduce()
        for i in range(n):
            for j in range(m):
                task.add_task(i, j, ops[i][j], prt[i][j])
        # for i, j in task.task_dict.items():
        #     print(i, j.machine, j.duration, "# operation(i,j),machine,duration")
        rule = JSPRule(n, m, task)
        spt = rule.spt
        lpt = rule.lpt
        assert spt == [0, 1, 2, 1, 0, 1, 0, 2, 2]  # 断言，判断与预期结果是否相符
        assert lpt == [2, 1, 0, 1, 2, 1, 2, 0, 0]  # 断言，判断与预期结果是否相符
        jsp = Jsp(n, m, ops, prt)
        info_spt = jsp.decode_jsp(np.array(spt))
        info_lpt = jsp.decode_jsp(np.array(lpt))
        info_spt.ganttChart_png(file_name="GanttChart-SPT-JSP-1")  # 保存在./Results/Figure目录下
        info_lpt.ganttChart_png(file_name="GanttChart-LPT-JSP-1")  # 保存在./Results/Figure目录下


if __name__ == "__main__":
    unittest.main()