Coverage for moptipy / examples / jssp / gantt_space.py: 79%
137 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-24 08:49 +0000
1"""Here we provide a :class:`~moptipy.api.space.Space` of `Gantt` charts."""
2from math import factorial
3from typing import Final
5import numpy as np
6from pycommons.types import check_int_range, type_error
8from moptipy.api.space import Space
9from moptipy.examples.jssp.gantt import Gantt
10from moptipy.examples.jssp.instance import SCOPE_INSTANCE, Instance
11from moptipy.utils.logger import CSV_SEPARATOR, KeyValueLogSection
12from moptipy.utils.nputils import (
13 KEY_NUMPY_TYPE,
14 int_range_to_dtype,
15 numpy_type_to_str,
16)
18#: the array shape
19KEY_SHAPE: Final[str] = "shape"
22def gantt_space_size(jobs: int, machines: int) -> int:
23 """
24 Compute the size of the Gantt space.
26 :param jobs: the number of jobs
27 :param machines: the number of machines
28 :return: the size of the search
30 >>> print(gantt_space_size(8, 5))
31 106562062388507443200000
32 """
33 return factorial(check_int_range(jobs, "jobs", 1, 1_000_000)) **\
34 check_int_range(machines, "machines", 1, 1_000_000)
37# start book
38class GanttSpace(Space):
39 """An implementation of the `Space` API of for `Gantt` charts."""
41 def __init__(self, instance: Instance) -> None:
42 # end book
43 """
44 Create a Gantt chart space.
46 :param instance: the JSSP instance
47 """
48 if not isinstance(instance, Instance):
49 raise type_error(instance, "instance", Instance)
50 #: The JSSP Instance to which the Gantt record apply.
51 self.instance: Final[Instance] = instance # +book
52 #: The shape for the Gantt chart arrays.
53 self.shape: Final[tuple[int, int, int]] = ( # +book
54 instance.machines, instance.jobs, 3) # +book
55 #: the data to be used for Gantt charts
56 self.dtype: Final[np.dtype] = int_range_to_dtype(
57 min_value=0, max_value=instance.makespan_upper_bound)
58 #: fast call function forward
59 self.copy = np.copyto # type: ignore # +book
61 def create(self) -> Gantt: # +book
62 """
63 Create a Gantt chart object without assigning jobs to machines.
65 :return: the Gantt chart
66 """
67 return Gantt(self) # +book
69 def to_str(self, x: Gantt) -> str: # +book
70 """
71 Convert a Gantt chart to a string.
73 :param x: the Gantt chart
74 :return: a string corresponding to the flattened `Gantt` array
75 """
76 return CSV_SEPARATOR.join(map(str, np.nditer(x))) # +book
78 def is_equal(self, x1: Gantt, x2: Gantt) -> bool: # +book
79 """
80 Check if two Gantt charts have the same contents.
82 :param x1: the first chart
83 :param x2: the second chart
84 :return: `True` if both charts are for the same instance and have the
85 same structure
86 """
87 # start book
88 return (x1.instance is x2.instance) and np.array_equal(x1, x2)
89 # end book
91 def from_str(self, text: str) -> Gantt: # +book
92 """
93 Convert a string to a Gantt chart.
95 :param text: the string
96 :return: the Gantt chart
97 """
98 if not isinstance(text, str):
99 raise type_error(text, "Gannt text", str)
100 # start book
101 x: Final[Gantt] = self.create()
102 np.copyto(x, np.fromstring(
103 text, dtype=self.dtype, sep=CSV_SEPARATOR)
104 .reshape(self.shape))
105 self.validate(x) # -book
106 return x
107 # end book
109 def validate(self, x: Gantt) -> None: # +book
110 """
111 Check if a Gantt chart is valid and feasible.
113 This means that the operations of the jobs must appear in the right
114 sequences and must not intersect in any way.
115 The only exception are operations that need 0 time units. They are
116 permitted to appear wherever.
118 :param x: the Gantt chart
119 :raises TypeError: if any component of the chart is of the wrong type
120 :raises ValueError: if the Gantt chart is not feasible or the makespan
121 is wrong
122 """
123 # start book
124 # Checks if a Gantt chart if valid and feasible.
125 if not isinstance(x, Gantt):
126 raise type_error(x, "x", Gantt)
127 # the rest of the checks is not printed for brevity reasons...
128 # end book
129 inst: Final[Instance] = self.instance
130 if inst is not x.instance:
131 raise ValueError(
132 f"x.instance must be {inst} but is {x.instance}.")
133 jobs: Final[int] = inst.jobs
134 machines: Final[int] = inst.machines
135 if len(x.shape) != 3:
136 raise ValueError("x must be three-dimensional, "
137 f"but is {len(x.shape)}-dimensional.")
138 if x.shape[0] != machines:
139 raise ValueError(
140 f"x must have {machines} rows for instance {inst.name}, "
141 f"but has {x.shape[0]}.")
143 if x.shape[1] != jobs:
144 raise ValueError(
145 f"x must have {jobs} columns for instance {inst.name}, "
146 f"but has {x.shape[1]}.")
147 if x.shape[2] != 3:
148 raise ValueError("x must have three values per cell, "
149 f"but has {x.shape[2]}.")
150 if x.dtype != self.dtype:
151 raise ValueError(
152 f"x.dtype should be {self.dtype} for instance "
153 f"{inst.name} but is {x.dtype}.")
155 # now check the sequence on operations per machine
156 # we check for overlaps and incorrect operation times
157 jobs_done: Final[set[int]] = set()
158 for machinei in range(machines):
159 jobs_done.clear()
160 last_end = 0
161 last_name = "[start]"
162 for jobi in range(jobs):
163 job = int(x[machinei, jobi, 0])
164 start = int(x[machinei, jobi, 1])
165 end = int(x[machinei, jobi, 2])
166 if not (0 <= job < jobs):
167 raise ValueError(
168 f"job {job} invalid for machine {machinei} on "
169 f"instance {inst.name} with {jobs} jobs: "
170 f"{x[machinei]}")
171 if job in jobs_done:
172 raise ValueError(
173 f"job {job} appears twice on machine {machinei} "
174 f"for instance {inst.name}: {x[machinei]}")
175 jobs_done.add(job)
176 time = -1
177 for z in range(machines):
178 if inst[job, z, 0] == machinei:
179 time = int(inst[job, z, 1])
180 break
181 if time < 0:
182 raise ValueError(
183 f"Did not find machine {machinei} for job {job} in "
184 f"instance {inst.name}: {x[machinei]}, "
185 f"{inst[machinei]}.")
186 if (end - start) != time:
187 raise ValueError(
188 f"job {job} should need {time} time units on machine "
189 f"{machinei} for instance {inst.name}, but starts at "
190 f"{start} and ends at {end}: {x[machinei]}, "
191 f"{inst[job]}.")
192 if time <= 0:
193 continue # job requires zero time, can skip
194 if start < last_end:
195 raise ValueError(
196 f"job {job} starts at {start} on machine {machinei}, "
197 f"for instance {inst.name} but cannot before "
198 f"{last_name}: {x[machinei]}.")
199 last_end = end
201 # now check the single jobs
202 # we again check for operation overlaps and incorrect timing
203 for jobi in range(jobs):
204 done = [(start, end, machine)
205 for machine in range(machines)
206 for (idx, start, end) in x[machine, :, :] if idx == jobi]
207 done.sort()
208 if len(done) != machines:
209 raise ValueError(
210 f"Job {jobi} appears only {len(done)} times instead of "
211 f"{machines} times on instance {inst.name}.")
213 # we allow operations of length 0 to appear at any position
214 last_end = 0
215 last_machine = "[start]"
216 done_i = 0
217 machine_i = 0
218 while True:
219 if machine_i < machines:
220 machine, time = inst[jobi, machine_i]
221 else:
222 machine = time = -1
223 if done_i < machines:
224 start = int(done[machine_i][0])
225 end = int(done[machine_i][1])
226 used_machine = int(done[machine_i][2])
227 else:
228 start = end = used_machine = -1
230 needed = end - start
231 if needed < 0:
232 raise ValueError(
233 f"Operation {machine_i} of job {jobi} scheduled "
234 f"from {start} to {end}?")
235 if machine != used_machine:
236 # This is only possible for operations that require zero
237 # time units. We will skip such operations in the checks.
238 if (time == 0) and (machine != -1):
239 machine_i += 1
240 continue
241 if (needed == 0) and (machine != -1):
242 done_i += 1
243 continue
244 raise ValueError(
245 f"Machine at index {done_i} of job {jobi} "
246 f"must be {machine} for instance {inst.name}, "
247 f"for {time} time units, but is {used_machine}"
248 f"from {start} to {end}.")
249 if machine == -1:
250 if (machine_i < machines) or (done_i < machines):
251 raise ValueError( # this should never be possible
252 f"Done {machine_i + 1} machines for job {jobi}, "
253 f"which has {done_i + 1} operations done??")
254 break # we can stop
256 # ok, we are at a regular operation
257 if needed != time:
258 raise ValueError(
259 f"Job {jobi} must be processed on {machine} for "
260 f"{time} time units on instance {inst.name}, but "
261 f"only {needed} are used from {start} to {end}.")
262 if needed < 0:
263 raise ValueError(
264 f"Processing time of job {jobi} on machine {machine} "
265 f"cannot be < 0 for instance {inst.name}, but "
266 f"is {needed}.")
268 if start < last_end:
269 raise ValueError(
270 f"Processing time window [{start},{end}] on "
271 f"machine {machine} for job {jobi} on instance "
272 f"{inst.name} intersects with last operation end"
273 f"{last_end} on machine {last_machine}.")
275 last_end = end
276 last_machine = machine
277 machine_i += 1
278 done_i += 1
280 maxtime: Final[int] = int(x[:, -1, 2].max())
281 if maxtime < inst.makespan_lower_bound:
282 raise ValueError(
283 f"Makespan {maxtime} computed, which is smaller than the "
284 f"lower bound {inst.makespan_lower_bound} of the JSSP "
285 f"instance {str(inst)!r}'.")
286 if maxtime > inst.makespan_upper_bound:
287 raise ValueError(
288 f"Makespan {maxtime} computed, which is larger than the "
289 f"upper bound {inst.makespan_upper_bound} of the JSSP "
290 f"instance {str(inst)!r}.")
292 def n_points(self) -> int:
293 """
294 Get the number of possible Gantt charts without useless delays.
296 :return: `factorial(jobs) ** machines`
298 >>> print(GanttSpace(Instance.from_resource("demo")).n_points())
299 7962624
300 """
301 return gantt_space_size(self.instance.jobs, self.instance.machines)
303 def __str__(self) -> str:
304 """
305 Get the name of the Gantt space.
307 :return: the name
309 >>> space = GanttSpace(Instance.from_resource("abz7"))
310 >>> print(space)
311 gantt_abz7
312 """
313 return f"gantt_{self.instance}"
315 def log_parameters_to(self, logger: KeyValueLogSection) -> None:
316 """
317 Log the parameters of the Gantt space to the given logger.
319 :param logger: the logger for the parameters
320 """
321 super().log_parameters_to(logger)
322 logger.key_value(KEY_SHAPE, repr(self.shape))
323 logger.key_value(KEY_NUMPY_TYPE, numpy_type_to_str(self.dtype))
324 with logger.scope(SCOPE_INSTANCE) as kv:
325 self.instance.log_parameters_to(kv)