added satisfy implementation for fjsp, added UT

discrete_optimization/fjsp/flex_job_shop_problem.py

@@ -78,11 +78,71 @@ def __init__(
                 )
                 for job in self.list_jobs
             )
+        self.nb_subjob_per_job = {
+            i: len(self.list_jobs[i].sub_jobs) for i in range(self.n_jobs)
+        }
+        self.subjob_possible_machines = {
+            (i, j): set(x.machine_id for x in self.list_jobs[i].sub_jobs[j])
+            for i in range(self.n_jobs)
+            for j in range(self.nb_subjob_per_job[i])
+        }
+        self.duration_per_machines = {
+            (i, j): {
+                x.machine_id: x.processing_time for x in self.list_jobs[i].sub_jobs[j]
+            }
+            for (i, j) in self.subjob_possible_machines
+        }
 
     def evaluate(self, variable: SolutionFJobshop) -> dict[str, float]:
         return {"makespan": max(x[-1][1] for x in variable.schedule)}
 
-    def satisfy(self, variable: Solution) -> bool:
+    def satisfy(self, variable: SolutionFJobshop) -> bool:
+        if not all(
+            variable.schedule[i][j][2] in self.subjob_possible_machines[(i, j)]
+            for (i, j) in self.subjob_possible_machines
+        ):
+            logger.info("Unallowed machine used for some subjob")
+            return False
+        for m in self.job_per_machines:
+            sorted_ = sorted(
+                [
+                    variable.schedule[x[0]][x[1]]
+                    for x in self.job_per_machines[m]
+                    if variable.schedule[x[0]][x[1]][2] == m
+                ],
+                key=lambda y: y[0],
+            )
+            len_ = len(sorted_)
+            for i in range(1, len_):
+                if sorted_[i][0] < sorted_[i - 1][1]:
+                    logger.info("Overlapping task on same machines")
+                    return False
+        for job in range(self.n_jobs):
+            m = variable.schedule[job][0][2]
+            if not (
+                variable.schedule[job][0][1] - variable.schedule[job][0][0]
+                == self.duration_per_machines[(job, 0)][m]
+            ):
+                logger.info(
+                    f"Duration of task {job, 0} not coherent with the machine choice "
+                )
+            for s_j in range(1, len(variable.schedule[job])):
+                if variable.schedule[job][s_j][0] < variable.schedule[job][s_j - 1][1]:
+                    logger.info(
+                        f"Precedence constraint not respected between {job, s_j}"
+                        f"and {job, s_j-1}"
+                    )
+                    return False
+                if not (
+                    variable.schedule[job][s_j][1] - variable.schedule[job][s_j][0]
+                    == self.duration_per_machines[(job, s_j)][
+                        variable.schedule[job][s_j][2]
+                    ]
+                ):
+                    logger.info(
+                        f"Duration of task {job, s_j} not coherent with the machine choice "
+                    )
+                    return False
         return True
 
     def get_attribute_register(self) -> EncodingRegister:

tests/fjsp/test_fjsp_cpsat.py

@@ -51,7 +51,7 @@ def test_cpsat_fjsp():
     p.nb_process = 10
     res = solver.solve(
         parameters_cp=p,
-        time_limit=300,
+        time_limit=30,
         ortools_cpsat_solver_kwargs=dict(log_search_progress=True),
         duplicate_temporal_var=True,
         add_cumulative_constraint=True,

tests/fjsp/test_fjsp_problem.py

@@ -0,0 +1,70 @@
+#  Copyright (c) 2024 AIRBUS and its affiliates.
+#  This source code is licensed under the MIT license found in the
+#  LICENSE file in the root directory of this source tree.
+
+import logging
+
+from discrete_optimization.fjsp.flex_job_shop_problem import (
+    Job,
+    SolutionFJobshop,
+    Subjob,
+)
+from discrete_optimization.fjsp.solvers.cpsat_fjsp_problem import (
+    CPSatFJspSolver,
+    FJobShopProblem,
+)
+
+logging.basicConfig(level=logging.INFO)
+
+
+def create_dummy_fjsp_and_sol():
+    job_0 = Job(
+        job_id=0,
+        sub_jobs=[
+            [Subjob(0, 1), Subjob(1, 2)],
+            [Subjob(3, 1), Subjob(4, 2)],
+            [Subjob(1, 1), Subjob(2, 2)],
+        ],
+    )
+    job_1 = Job(
+        job_id=1,
+        sub_jobs=[
+            [Subjob(0, 1), Subjob(1, 2)],
+            [Subjob(3, 1), Subjob(4, 2)],
+            [Subjob(1, 4), Subjob(2, 2)],
+        ],
+    )
+    problem = FJobShopProblem(
+        list_jobs=[job_0, job_1], n_jobs=2, n_machines=5, horizon=30
+    )
+    sol = SolutionFJobshop(
+        problem=problem,
+        schedule=[[(0, 1, 0), (1, 2, 3), (2, 4, 2)], [(0, 2, 1), (2, 4, 4), (4, 6, 2)]],
+    )
+    return problem, sol
+
+
+def test_fjsp_satisfy():
+    problem, sol = create_dummy_fjsp_and_sol()
+    assert problem.satisfy(sol)
+    # Overlap of machine 0 at time 0 !
+    sol = SolutionFJobshop(
+        problem=problem,
+        schedule=[[(0, 1, 0), (1, 2, 3), (2, 4, 2)], [(0, 1, 0), (2, 4, 4), (4, 6, 2)]],
+    )
+    assert not problem.satisfy(sol)
+
+    # Wrong machine on job (1, 0)
+
+    sol = SolutionFJobshop(
+        problem=problem,
+        schedule=[[(0, 1, 0), (1, 2, 3), (2, 4, 2)], [(0, 2, 5), (2, 4, 4), (4, 6, 2)]],
+    )
+    assert not problem.satisfy(sol)
+
+    # Precedence constraint broken between (1, 2) and (1, 1)
+    sol = SolutionFJobshop(
+        problem=problem,
+        schedule=[[(0, 1, 0), (1, 2, 3), (2, 4, 2)], [(0, 2, 1), (4, 6, 4), (2, 4, 2)]],
+    )
+    assert not problem.satisfy(sol)