Source code for quantum_launcher.problems.problem_formulations.jssp.scheduler


[docs]
 1def get_label(task, time):
 2    return f"{task.job}_{task.position},{time}"

 3
 4

[docs]
 5class Task:
 6    def __init__(self, job, position, machine, duration):
 7        self.job = job
 8        self.position = position
 9        self.machine = machine
10        self.duration = duration
11
12    def __repr__(self):
13        return ("{{job: {job}, position: {position}, machine: {machine}, duration:"
14                " {duration}}}").format(**vars(self))

15
16

[docs]
17class KeyList:
18
19    def __init__(self, array, key_function):
20        self.array = array
21        self.key_function = key_function
22
23    def __len__(self):
24        return len(self.array)
25
26    def __getitem__(self, index):
27        item = self.array[index]
28        key = self.key_function(item)
29        return key

30
31

[docs]
32class JobShopScheduler:
33    def __init__(self, job_dict, max_time=None):
34        self.tasks = []
35        self.last_task_indices = []
36        self.max_time = max_time
37        self.H = 0
38        self.H_vars = {}
39        self.absurd_times = set()
40        self._process_data(job_dict)
41
42    def _process_data(self, jobs):
43        tasks = []
44        last_task_indices = [-1]
45        total_time = 0
46
47        for job_name, job_tasks in jobs.items():
48            last_task_indices.append(last_task_indices[-1] + len(job_tasks))
49
50            for i, (machine, time_span) in enumerate(job_tasks):
51                tasks.append(Task(job_name, i, machine, time_span))
52                total_time += time_span
53
54        self.tasks = tasks
55        self.last_task_indices = last_task_indices[1:]
56
57        if self.max_time is None:
58            self.max_time = total_time
59
60    def _remove_absurd_times(self, disable_till: dict, disable_since, disabled_variables):
61        predecessor_time = 0
62        current_job = self.tasks[0].job
63        for task in self.tasks:
64            if task.job != current_job:
65                predecessor_time = 0
66                current_job = task.job
67
68            for t in range(predecessor_time):
69                label = get_label(task, t)
70                self.absurd_times.add(label)
71
72            predecessor_time += task.duration
73
74        successor_time = -1
75        current_job = self.tasks[-1].job
76        for task in self.tasks[::-1]:
77            if task.job != current_job:
78                successor_time = -1
79                current_job = task.job
80
81            successor_time += task.duration
82            for t in range(successor_time):
83                label = get_label(task, (self.max_time - 1) - t)
84                self.absurd_times.add(label)
85
86        for task in self.tasks:
87            if task.machine in disable_till.keys():
88                for i in range(disable_till[task.machine]):
89                    label = get_label(task, i)
90                    self.absurd_times.add(label)
91            elif task.machine in disable_since.keys():
92                for i in range(disable_since[task.machine], self.max_time):
93                    label = get_label(task, i)
94                    self.absurd_times.add(label)
95
96        for label in disabled_variables:
97            self.absurd_times.add(label)