'Refactored by Sourcery'

benchmark/benchmark_dev_team.py

@@ -85,12 +85,12 @@ def get_size(byt, suffix="B"):
 
 N = list(range(10, n, step))  # from 4 to N, step 2
 
+# Teams and Resources
+num_resource_a = 2
+num_resource_b = 2
+
 for num_dev_teams in N:
     print("-> Num dev teams:", num_dev_teams)
-    # Teams and Resources
-    num_resource_a = 2
-    num_resource_b = 2
-
     init_time = time.perf_counter()
     # Resources
     digital_transformation = ps.SchedulingProblem(

benchmark/benchmark_mixed.py

@@ -85,20 +85,18 @@ def get_size(byt, suffix="B"):
 computation_times = []
 plot_abs = []
 
+MAX_TASKS_PER_PERIOD = 2
+MAX_TASKS_IN_PROBLEM = 4
+NB_WORKERS = 10
+NB_TASKS_PER_WORKER = 10
 for horizon in range(20, n, step):
-    MAX_TASKS_PER_PERIOD = 2
-    MAX_TASKS_IN_PROBLEM = 4
-    NB_WORKERS = 10
-    NB_TASKS_PER_WORKER = 10
     PERIODS = [
         (10 * i, 10 * (i + 1)) for i in range(int(horizon / 10))
     ]  # Periods of 10 slots from 0 to horizon
     init_time = time.perf_counter()
 
     # Create problem and initialize constraints
     pb = ps.SchedulingProblem(name="performance_analyzer", horizon=horizon)
-    constraints = []
-
     # Create resources and assign tasks
     general_worker = ps.Worker("general")
     workers = []
@@ -118,8 +116,11 @@ def get_size(byt, suffix="B"):
 
     workload = {period: MAX_TASKS_PER_PERIOD for period in PERIODS}
     workload[(0, horizon)] = MAX_TASKS_IN_PROBLEM
-    for worker in workers:
-        constraints.append(ps.WorkLoad(worker["worker"], workload, kind="max"))
+    constraints = [
+        ps.WorkLoad(worker["worker"], workload, kind="max")
+        for worker in workers
+    ]
+
 
     # Add constraints, define objective and solve problem
     pb.add_constraints(constraints)

processscheduler/base.py

@@ -75,7 +75,7 @@ def __repr__(self) -> str:
             type(self),
             len(self.assertions),
         )
-        assertions_str = "".join(["%s" % ass for ass in self.assertions])
+        assertions_str = "".join("%s" % ass for ass in self.assertions)
         return str_to_return + assertions_str
 
     def add_assertion(self, z3_assertion: BoolRef) -> bool:
@@ -149,9 +149,9 @@ def __init__(
                 )
 
         # all scheduled variables to take into account
-        applied_vars = []
-        for constraint in list_of_optional_constraints:
-            applied_vars.append(constraint.applied)
+        applied_vars = [
+            constraint.applied for constraint in list_of_optional_constraints
+        ]
 
         asst = problem_function[kind](
             [(applied, True) for applied in applied_vars], nb_constraints_to_apply

processscheduler/cost.py

@@ -37,10 +37,7 @@ def plot(self, interval, show_plot=True) -> None:
 
         lower_bound, upper_bound = interval
         x = np.linspace(lower_bound, upper_bound, 1000)
-        y = []
-        for x_ in x:
-            y.append(self.f(x_))
-
+        y = [self.f(x_) for x_ in x]
         plt.plot(x, y, label="Cost function")
 
         plt.legend()

processscheduler/jupyter_ui.py

@@ -230,9 +230,10 @@ def on_create_resource_button_clicked(b):
             cost=ConstantCostPerPeriod(resource_cost_per_period_widget.value),
         )
     # rebuild option list for the task list
-    resources_list_of_tuples = []
-    for resource in pb.context.resources:
-        resources_list_of_tuples.append((resource.name, resource))
+    resources_list_of_tuples = [
+        (resource.name, resource) for resource in pb.context.resources
+    ]
+
     resources_select_widget.options = resources_list_of_tuples
     with resource_output:
         print("Resource", new_resource.name, "successfully created.")
@@ -281,9 +282,7 @@ def on_change_task_type(change):
         new_value = "%s" % change["new"]
         if new_value == "FixedDurationTask":
             task_duration_widget.disabled = False
-        elif new_value == "ZeroDurationTask":
-            task_duration_widget.disabled = True
-        elif new_value == "VariableDurationTask":
+        elif new_value in ["ZeroDurationTask", "VariableDurationTask"]:
             task_duration_widget.disabled = True
 
 
@@ -340,9 +339,7 @@ def on_create_task_button_clicked(b):
             optional=task_optional,
         )
     # rebuild option list for the task list
-    tasks_list_of_tuples = []
-    for task in pb.context.tasks:
-        tasks_list_of_tuples.append((task.name, task))
+    tasks_list_of_tuples = [(task.name, task) for task in pb.context.tasks]
     tasks_select_widget.options = tasks_list_of_tuples
     with task_output:
         print("Task", task_name, "successfully created.")
@@ -443,10 +440,11 @@ def assign_all_workers_resource_button_clicked(b) -> bool:
         selected_task.add_required_resources(selected_resources)
         print(
             "Assign",
-            ",".join([s.name for s in selected_resources]),
+            ",".join(s.name for s in selected_resources),
             "to task",
             selected_task.name,
         )
+
     return True
 
 
@@ -499,10 +497,11 @@ def assign_alternative_workers_resource_button_clicked(b) -> bool:
         print(
             "Assign %i (%s) resources among"
             % (nb_workers_widget.value, select_worker_type_widget.value),
-            ",".join([s.name for s in selected_resources]),
+            ",".join(s.name for s in selected_resources),
             "to task",
             selected_task.name,
         )
+
         return True
 
 

processscheduler/problem.py

@@ -92,13 +92,11 @@ def add_indicator_number_tasks_assigned(self, resource: _Resource):
         ]
 
         nb_tasks_assigned_indicator_variable = Sum(scheduled_tasks)
-        nb_tasks_assigned_indicator = Indicator(
+        return Indicator(
             "Nb Tasks Assigned (%s)" % resource.name,
             nb_tasks_assigned_indicator_variable,
         )
 
-        return nb_tasks_assigned_indicator
-
     def add_indicator_resource_cost(
         self, list_of_resources: List[_Resource]
     ) -> Indicator:
@@ -148,10 +146,9 @@ def add_indicator_resource_utilization(self, resource: _Resource) -> Indicator:
             for interv_low, interv_up in resource.busy_intervals.values()
         ]
         utilization = (Sum(durations) * 100) / self.horizon  # in percentage
-        utilization_indicator = Indicator(
+        return Indicator(
             "Utilization (%s)" % resource.name, utilization, bounds=(0, 100)
         )
-        return utilization_indicator
 
     def maximize_indicator(self, indicator: Indicator) -> MaximizeObjective:
         """Maximize indicator"""
@@ -269,14 +266,10 @@ def add_objective_flowtime_single_resource(
             "GreatestTaskEndTimeInTimePeriodForResource%s_%s" % (resource.name, uid)
         )
 
-        asst_max = []
-        for task in resource.busy_intervals:
-            asst_max.append(
-                Implies(
+        asst_max = [Implies(
                     And(task.end <= upper_bound, task.start >= lower_bound),
                     maxi == task.end,
-                )
-            )
+                ) for task in resource.busy_intervals]
         flowtime_single_resource_indicator.add_assertion(Or(asst_max))
         for task in resource.busy_intervals:
             flowtime_single_resource_indicator.add_assertion(
@@ -291,14 +284,10 @@ def add_objective_flowtime_single_resource(
             "SmallestTaskEndTimeInTimePeriodForResource%s_%s" % (resource.name, uid)
         )
 
-        asst_min = []
-        for task in resource.busy_intervals:
-            asst_min.append(
-                Implies(
+        asst_min = [Implies(
                     And(task.end <= upper_bound, task.start <= lower_bound),
                     mini == task.start,
-                )
-            )
+                ) for task in resource.busy_intervals]
         flowtime_single_resource_indicator.add_assertion(Or(asst_min))
         for task in resource.busy_intervals:
             flowtime_single_resource_indicator.add_assertion(

processscheduler/solution.py

@@ -25,7 +25,7 @@
 
 class SolutionJSONEncoder(json.JSONEncoder):
     def default(self, obj):
-        if isinstance(obj, datetime) or isinstance(obj, (time, timedelta)):
+        if isinstance(obj, (datetime, time, timedelta)):
             return "%s" % obj
         return obj.__dict__
 
@@ -80,20 +80,20 @@ def __repr__(self):
     def get_all_tasks_but_unavailable(self):
         """Return all tasks except those of the type UnavailabilityTask
         used to represent a ResourceUnavailable constraint."""
-        tasks_to_return = {}
-        for task in self.tasks:
-            if "NotAvailable" not in task:
-                tasks_to_return[task] = self.tasks[task]
-        return tasks_to_return
+        return {
+            task: self.tasks[task]
+            for task in self.tasks
+            if "NotAvailable" not in task
+        }
 
     def get_scheduled_tasks(self):
         """Return scheduled tasks."""
         tasks_not_unavailable = self.get_all_tasks_but_unavailable()
-        tasks_to_return = {}
-        for task in tasks_not_unavailable:
-            if tasks_not_unavailable[task].scheduled:
-                tasks_to_return[task] = tasks_not_unavailable[task]
-        return tasks_to_return
+        return {
+            task: tasks_not_unavailable[task]
+            for task in tasks_not_unavailable
+            if tasks_not_unavailable[task].scheduled
+        }
 
     def to_json_string(self) -> str:
         """Export the solution to a json string."""
@@ -103,20 +103,19 @@ def to_json_string(self) -> str:
         d["horizon"] = self.horizon
         # time data
         problem_properties["problem_timedelta"] = self.problem.delta_time
-        if self.problem.delta_time is not None:
-            if self.problem.start_time is not None:
-                problem_properties["problem_start_time"] = self.problem.start_time
-                problem_properties["problem_end_time"] = (
-                    self.problem.start_time + self.horizon * self.problem.delta_time
-                )
-            else:
-                problem_properties["problem_start_time"] = time(0)
-                problem_properties["problem_end_time"] = (
-                    self.horizon * self.problem.delta_time
-                )
-        else:
+        if self.problem.delta_time is None:
             problem_properties["problem_start_time"] = None
             problem_properties["problem_end_time"] = None
+        elif self.problem.start_time is not None:
+            problem_properties["problem_start_time"] = self.problem.start_time
+            problem_properties["problem_end_time"] = (
+                self.problem.start_time + self.horizon * self.problem.delta_time
+            )
+        else:
+            problem_properties["problem_start_time"] = time(0)
+            problem_properties["problem_end_time"] = (
+                self.horizon * self.problem.delta_time
+            )
         d["problem_properties"] = problem_properties
 
         d["tasks"] = self.tasks
@@ -189,10 +188,7 @@ def render_gantt_plotly(
                 )
 
         r = lambda: random.randint(0, 255)
-        colors = []
-        for _ in range(len(df)):
-            colors.append("#%02X%02X%02X" % (r(), r(), r()))
-
+        colors = ["#%02X%02X%02X" % (r(), r(), r()) for _ in df]
         if sort is not None:
             if sort in ["Task", "Resource"]:
                 df = sorted(df, key=lambda i: i[sort], reverse=False)
@@ -333,10 +329,7 @@ def render_gantt_matplotlib(
 
         def draw_broken_barh_with_text(start, length, bar_color, text, hatch=None):
             # first compute the bar dimension
-            if length == 0:  # zero duration tasks, to be visible
-                bar_dimension = (start - 0.05, 0.1)
-            else:
-                bar_dimension = (start, length)
+            bar_dimension = (start - 0.05, 0.1) if length == 0 else (start, length)
             gantt.broken_barh(
                 [bar_dimension],
                 (i * 2, 2),

processscheduler/solver.py

@@ -264,17 +264,13 @@ def build_solution(self, z3_sol):
                         self._problem.start_time
                         + new_task_solution.start * self._problem.delta_time
                     )
-                    new_task_solution.end_time = (
-                        new_task_solution.start_time + new_task_solution.duration_time
-                    )
                 else:
                     new_task_solution.start_time = (
                         new_task_solution.start * self._problem.delta_time
                     )
-                    new_task_solution.end_time = (
-                        new_task_solution.start_time + new_task_solution.duration_time
-                    )
-
+                new_task_solution.end_time = (
+                    new_task_solution.start_time + new_task_solution.duration_time
+                )
             if task.optional:
                 # ugly hack, necessary because there's no as_bool()
                 # method for Bool objects
@@ -284,17 +280,12 @@ def build_solution(self, z3_sol):
 
             # process resource assignments
             for req_res in task.required_resources:
-                # by default, resource_should_be_assigned is set to True
-                # if will be set to False if the resource is an alternative worker
-                resource_is_assigned = True
                 # among those workers, some of them
                 # are busy "in the past", that is to say they
                 # should not be assigned to the related task
                 # for each interval
                 lower_bound, _ = req_res.busy_intervals[task]
-                if z3_sol[lower_bound].as_long() < 0:
-                    # should not be scheduled
-                    resource_is_assigned = False
+                resource_is_assigned = z3_sol[lower_bound].as_long() >= 0
                 # add this resource to assigned resources, anytime
                 if resource_is_assigned and (
                     req_res.name not in new_task_solution.assigned_resources
@@ -433,12 +424,11 @@ def solve_optimize_incremental(
 
         while True:  # infinite loop, break if unsat of max_depth
             depth += 1
-            if max_recursion_depth is not None:
-                if depth > max_recursion_depth:
-                    warnings.warn(
-                        "maximum recursion depth exceeded. There might be a better solution."
-                    )
-                    break
+            if max_recursion_depth is not None and depth > max_recursion_depth:
+                warnings.warn(
+                    "maximum recursion depth exceeded. There might be a better solution."
+                )
+                break
 
             is_sat, sat_computation_time = self.check_sat()
 
@@ -447,7 +437,7 @@ def solve_optimize_incremental(
                     "\tFound optimum %i. Stopping iteration." % current_variable_value
                 )
                 break
-            elif is_sat == unsat and current_variable_value is None:
+            elif is_sat == unsat:
                 print("\tNo solution found. Stopping iteration.")
                 break
             elif is_sat == unknown:

processscheduler/task_constraint.py

@@ -54,10 +54,7 @@ def __init__(
         self.offset = offset
         self.kind = kind
 
-        if offset > 0:
-            lower = task_before.end + offset
-        else:
-            lower = task_before.end
+        lower = task_before.end + offset if offset > 0 else task_before.end
         upper = task_after.start
 
         if kind == "lax":
@@ -280,10 +277,7 @@ def __init__(
                     "This class %s must excplicitely be set as optional." % task.name
                 )
         # all scheduled variables to take into account
-        sched_vars = []
-        for task in list_of_optional_tasks:
-            sched_vars.append(task.scheduled)
-
+        sched_vars = [task.scheduled for task in list_of_optional_tasks]
         asst = problem_function[kind](
             [(scheduled, True) for scheduled in sched_vars], nb_tasks_to_schedule
         )