1from abc import ABC, abstractmethod
2from dataclasses import dataclass
3from typing import Literal
4
5from pyqubo import Binary, Model # type: ignore
6
7from qlauncher import hampy
8from qlauncher.hampy import Equation, Variable
9
10
[docs]
11@dataclass
12class Task:
13 uid: int
14 job: str
15 position: int
16 machine: str
17 duration: int
18
19 def __repr__(self):
20 return ('{{job: {job}, position: {position}, machine: {machine}, duration: {duration}}}').format(**vars(self))
21
22 def __hash__(self) -> int:
23 return self.uid.__hash__()
24
25
[docs]
26class JobShopScheduler(ABC):
27 tasks: list[Task]
28 tasks_by_machine: dict[str, set[Task]]
29 tasks_by_job: dict[str, list[Task]]
30 max_time: int
31 valid_assignments: set[tuple[Task, int]]
32 n: int
33
34 def __init__(
35 self,
36 job_dict: dict[str, list[tuple[str, int]]],
37 max_time: int | None = None,
38 ):
39 self.tasks = []
40 self.tasks_by_machine = {}
41 self.tasks_by_job = {}
42 self.valid_assignments = set()
43 self._process_data(job_dict, max_time)
44 self._prepare_valid_assignments()
45 self.n = len(self.valid_assignments)
46
47 def _process_data(self, jobs: dict[str, list[tuple[str, int]]], max_time: int | None = None) -> None:
48 tasks = []
49 last_task_indices = [-1]
50 total_time = 0
51
52 for job_name, job_tasks in jobs.items():
53 last_task_indices.append(last_task_indices[-1] + len(job_tasks))
54
55 for i, (machine, time_span) in enumerate(job_tasks):
56 task = Task(len(tasks), job_name, i, machine, time_span)
57 tasks.append(task)
58 self.tasks_by_job.setdefault(job_name, []).append(task)
59 self.tasks_by_machine.setdefault(job_name, set()).add(task)
60 total_time += time_span
61
62 self.tasks = tasks
63
64 if max_time is not None:
65 self.max_time = max_time
66 else:
67 self.max_time = total_time
68
69 def _prepare_valid_assignments(self) -> None:
70 for task in self.tasks:
71 for t in range(self.max_time):
72 self.valid_assignments.add((task, t))
73 self._exclude_invalid()
74 self.assignment_index = {}
75 i = 0
76 for task in self.tasks:
77 for t in range(self.max_time):
78 if self.valid(task, t):
79 self.assignment_index[(task, t)] = i
80 i += 1
81
82 def _exclude_invalid(self) -> None:
83 for tasks in self.tasks_by_job.values():
84 predecessor_time = 0
85 for task in tasks:
86 for t in range(predecessor_time):
87 if (task, t) in self.valid_assignments:
88 self.valid_assignments.remove((task, t))
89 predecessor_time += task.duration
90
91 successor_time = -1
92 for task in tasks[::-1]:
93 successor_time += task.duration
94 for t in range(successor_time):
95 if (task, t) in self.valid_assignments:
96 self.valid_assignments.remove((task, t))
97
98 def _exclude_on_demand(
99 self, disable_till: dict[str, int], disable_since: dict[str, int], disabled_variables: list[tuple[str, int, int]]
100 ) -> None:
101 for task in self.tasks:
102 if task.machine in disable_till:
103 for i in range(disable_till[task.machine]):
104 self.valid_assignments.remove((task, i))
105 elif task.machine in disable_since:
106 for i in range(disable_since[task.machine], self.max_time):
107 self.valid_assignments.remove((task, i))
108
109 for job, position, time in disabled_variables:
110 task = next(filter(lambda x: x.job == job and x.position == position, self.tasks))
111 self.valid_assignments.remove((task, time))
112
[docs]
113 def valid(self, task: Task, time: int) -> bool:
114 return (task, time) in self.valid_assignments
115
116 @abstractmethod
117 def _get_variable(self, task: Task, time: int) -> Binary | hampy.Variable | None:
118 pass
119
120 @abstractmethod
121 def _add_expression(self, var1: Binary | Variable, var2: Binary | Variable, lagrange_factor: float) -> None:
122 pass
123
124 @abstractmethod
125 def _add_expression_one_start(self, variables: list[Binary] | list[int | Variable], lagrange_factor: float) -> None:
126 pass
127
128 @abstractmethod
129 def _add_variable(self, var: Variable | Binary, bias: float) -> None:
130 pass
131
132 @abstractmethod
133 def _get_final(self) -> Equation | tuple[Model, int] | None:
134 pass
135
136 def _add_one_start_constraint(self, lagrange_one_hot: float = 1) -> None:
137 """self.csp gets the constraint: A task can start once and only once"""
138 for task in self.tasks:
139 variables = []
140 for t in range(self.max_time):
141 if not self.valid(task, t):
142 continue
143 variables.append(self._get_variable(task, t))
144 self._add_expression_one_start(variables, lagrange_one_hot)
145
146 def _add_precedence_constraint(self, lagrange_precedence: float = 1) -> None:
147 """self.csp gets the constraint: Task must follow a particular order.
148 Note: assumes self.tasks are sorted by jobs and then by position
149 """
150 for tasks in self.tasks_by_job.values():
151 for current_task, next_task in zip(tasks, tasks[1:], strict=False):
152 # Forming constraints with the relevant times of the next task
153 for t in range(self.max_time):
154 if not self.valid(current_task, t):
155 continue
156 var1 = self._get_variable(current_task, t)
157 for tt in range(min(t + current_task.duration, self.max_time)):
158 if not self.valid(next_task, tt):
159 continue
160 var2 = self._get_variable(next_task, tt)
161 self._add_expression(var1, var2, lagrange_precedence)
162
163 def _add_share_machine_constraint(self, lagrange_share: float = 1) -> None:
164 """self.csp gets the constraint: At most one task per machine per time unit"""
165 for tasks in self.tasks_by_machine.values():
166 # Apply constraint between all tasks for each unit of time
167 for task1 in tasks:
168 for task2 in tasks - {task1}:
169 for t in range(self.max_time):
170 if not self.valid(task1, t):
171 continue
172 var1 = self._get_variable(task1, t)
173 for tt in range(t, min(t + task1.duration, self.max_time)):
174 if not self.valid(task2, tt):
175 continue
176 var2 = self._get_variable(task2, tt)
177 self._add_expression(var1, var2, lagrange_share)
178
[docs]
179 def get_result(
180 self,
181 lagrange_one_hot: float,
182 lagrange_precedence: float,
183 lagrange_share: float,
184 version: Literal['decision', 'optimization'] = 'optimization',
185 ) -> Equation | Model:
186 self._add_one_start_constraint(lagrange_one_hot)
187 self._add_precedence_constraint(lagrange_precedence)
188 self._add_share_machine_constraint(lagrange_share)
189
190 base = len(self.tasks_by_job)
191
192 # Get BQM
193 if version == 'decision':
194 final = self._get_final()
195 if final is None:
196 raise TypeError
197 return final
198
199 for tasks in self.tasks_by_job.values():
200 task = tasks[-1]
201
202 for t in range(self.max_time):
203 end_time = t + task.duration
204 bias = 2 * base ** (end_time - self.max_time)
205 if not self.valid(task, t):
206 continue
207
208 var = self._get_variable(task, t)
209 self._add_variable(var, bias)
210
211 # Get BQM
212 final = self._get_final()
213 if final is None:
214 raise TypeError
215 return final
