linting

AGV_quantum/LinearProg.py

@@ -90,7 +90,7 @@ def _to_bqm_qubo_ising(self, pdict=None):
         pyqubo_model = H.compile()
         if pdict is None:
             pdict = {f"eq_{i}": 2 for i in range(self.num_eq)}
-        elif isinstance(pdict, int) or isinstance(pdict, float):
+        elif isinstance(pdict, (int, float)):
             pdict = {f"eq_{i}": pdict for i in range(self.num_eq)}
         pdict["obj"] = 1
         self.qubo = pyqubo_model.to_qubo(feed_dict=pdict)
@@ -123,11 +123,11 @@ def interpreter(sampleset: dimod.SampleSet):
         self.interpreter = lambda ss: interpreter(ss)
 
     @staticmethod
-    def _get_slack_ub(vars: list, coefs: list, offset: int) -> int:
+    def _get_slack_ub(problem_vars: list, coefs: list, offset: int) -> int:
         """Returns upper bound for slack variables
 
-        :param vars: List of variables (can be integer or binary)
-        :type vars: list
+        :param problem_vars: List of variables (can be integer or binary)
+        :type problem_vars: list
         :param coefs: List of coefficients for the inequality
         :type coefs: list
         :param offset: RHS of the inequality
@@ -136,7 +136,7 @@ def _get_slack_ub(vars: list, coefs: list, offset: int) -> int:
         :rtype: int
         """
         ub = 0
-        for var, coef in zip(vars, coefs):
+        for var, coef in zip(problem_vars, coefs):
             if isinstance(var, LogEncInteger):
                 ub += coef * (var.value_range[1] if coef < 0 else var.value_range[0])
             else:
@@ -192,7 +192,7 @@ def _to_cqm(self):
 
     def _count_qubits(self):
         model_vars = self.bqm.variables
-        return len(model_vars) 
+        return len(model_vars)
 
     def _count_quadratic_couplings(self):
         """
@@ -207,11 +207,9 @@ def _count_quadratic_couplings(self):
     def _count_linear_fields(self):
         """
         return number of local fields hs
-        """ 
+        """
         count = 0
         for h in self.bqm.linear.values():
             if h != 0:
                 count = count + 1
-        return count             
-
-
+        return count

AGV_quantum/__init__.py

@@ -8,7 +8,7 @@
 
 from .utils import(
     see_non_zero_variables, create_graph, create_iterators, create_agv_list, check_solution_list, qubo_to_matrix,
-    create_stations_list, create_agv_list, agv_routes_as_edges, create_same_way_dict, create_v_in_out,
+    create_stations_list, agv_routes_as_edges, create_same_way_dict, create_v_in_out,
     create_t_iterator, create_y_iterator, create_z_iterator
 )
 
@@ -23,4 +23,3 @@
 )
 
 from .train_diagram import plot_train_diagram, get_number_zones, zones_location, AGVS_coordinates
-

AGV_quantum/linear_solver.py

@@ -152,7 +152,7 @@ def _create_precedence_matrix_y(self):
         for y1, y2 in itertools.combinations(self.y_iter, r=2):
             if y1[2] == y2[2] and y1[0] == y2[1] and y1[1] == y2[0]:
                 t_vect = [0 for _ in self.t_iter]
-                y_vect = [1 if (y == y1 or y == y2) else 0 for y in self.y_iter]
+                y_vect = [1 if y in (y1, y2) else 0 for y in self.y_iter]
                 z_vect = [0 for _ in self.z_iter]
                 PVY.append(t_vect + y_vect + z_vect)
                 PVY_b.append(1)
@@ -171,7 +171,7 @@ def _create_precedence_matrix_z(self):
                 if z1[0] == z2[1] and z1[1] == z2[0] and z1[2] == z2[3] and z1[3] == z2[2]:
                     t_vect = [0 for _ in self.t_iter]
                     y_vect = [0 for _ in self.y_iter]
-                    z_vect = [1 if z == z1 or z == z2 else 0 for z in self.z_iter]
+                    z_vect = [1 if z in (z1, z2) else 0 for z in self.z_iter]
                     PVZ.append(t_vect + y_vect + z_vect)
                     PVZ_b.append(1)
 
@@ -234,7 +234,6 @@ def _create_minimal_headway_matrix(self, M: int, tracks, tau_headway: dict):
 
         MH = []
         MH_b = []
-        # TODO add condition s*
         for j1, j2, s, sp in tau_headway.keys():
             t_in_vect = [0 for _ in self.t_in_iter]
             t_out_vect = [1 if t == ("out", j1, s) else -1 if t == ("out", j2, s) else 0 for t in

AGV_quantum/process_results.py

@@ -120,4 +120,4 @@ def process_result(sampleset: list):
                 broken_constrains.append(eq)
     num_broken = len(broken_constrains)
 
-    return {"energy": energy, "objective": objective, "feasible": feasible, "num_broken": num_broken}
+    return {"energy": energy, "objective": objective, "feasible": feasible, "num_broken": num_broken}

AGV_quantum/qubo_solver.py

@@ -127,7 +127,7 @@ def get_file_name(
     """
     folder = os.path.join("annealing_results", input_name)
     fname = f"{method}"
-    if num_reads != None:
+    if num_reads is not None:
         fname += f"_{num_reads}_{annealing_time}_{chain_strength}"
     return os.path.join(folder, fname)
 
@@ -242,4 +242,3 @@ def annealing(
     if method != "cqm":
         sampleset = lp.interpreter(sampleset)
     return get_results(sampleset, prob=lp)
-

AGV_quantum/train_diagram.py

@@ -14,16 +14,16 @@ def get_number_zones(track_len):
         for el in tup:
             if el not in zones:
                 zones.append(el)
-    return len(zones)    
+    return len(zones)
 
 
 
 def zones_location(track_len, n_zones, s_ofset):
     """determines zones locations and borders in space for the plot"""
-    marks = {f"s{k}":0 for k in range(n_zones)} 
+    marks = {f"s{k}":0 for k in range(n_zones)}
     zone_borders = []
     prev_s = -100
-    x = 0 
+    x = 0
     for s in marks:
         if prev_s != -100:
             x = x + s_ofset + track_len[(f"{prev_s}", f"{s}")]
@@ -56,7 +56,7 @@ def plot_train_diagram(sol, agv_routes, track_len):
     marks, zone_borders = zones_location(track_len, n_zones, s_ofset)
     times, spaces = AGVS_coordinates(sol, agv_routes, marks, s_ofset)
     colors = {0: "black", 1: "red", 2: "green", 3: "blue", 4: "orange", 5: "brown", 6: "cyan"}
-    
+
     for agv in agv_routes:
         if n_zones < 8:
             plt.plot(times[agv], spaces[agv], "o-", label=f"$AGV_{agv}$ ", color=colors[agv], linewidth=0.85, markersize=2)
@@ -67,11 +67,11 @@ def plot_train_diagram(sol, agv_routes, track_len):
     for el in zone_borders:
         plt.axhline(y = el, color="gray", linewidth=0.5, linestyle=":")
         locs = [marks[k] for k in marks]
-    
+
     our_marks = [f"$s_{i}$" for i, _ in enumerate(marks) ]
 
     plt.yticks(locs, our_marks)
     plt.xlabel("time")
     plt.ylabel("zones")
-    plt.subplots_adjust(bottom=0.19, top = 0.75)        
+    plt.subplots_adjust(bottom=0.19, top = 0.75)
     plt.savefig("train_diagram.pdf")

AGV_quantum/utils.py

@@ -1,12 +1,11 @@
-#from collections import OrderedDict
+#helpers
 import itertools
 import os
 import networkx as nx
 import pandas as pd
 import numpy as np
-from docplex.mp.model_reader import ModelReader
-from AGV_quantum import LinearProg
 import dimod
+from AGV_quantum import LinearProg
 from AGV_quantum import get_results
 from AGV_quantum import process_result
 
@@ -23,7 +22,7 @@ def create_stations_list(tracks):
     return list(set(stations))
 
 
-def agv_routes_as_edges(agv_routes): 
+def agv_routes_as_edges(agv_routes):
     """ """
     return_dict = {}
     for j in agv_routes.keys():
@@ -102,7 +101,7 @@ def create_z_iterator(graph: nx.Graph, agv_routes):
     agv_routes_as_edges_dict = agv_routes_as_edges(agv_routes)
 
     for j1, j2 in list(itertools.permutations(J, r=2)):
-        if j1 in agv_routes_as_edges_dict.keys() and j2 in agv_routes_as_edges_dict.keys():
+        if j1 in agv_routes_as_edges_dict and j2 in agv_routes_as_edges_dict:
             for s, sp in agv_routes_as_edges_dict[j1]:
                 if graph.number_of_edges(s, sp) < 2 and (sp, s) in agv_routes_as_edges_dict[j2]:
                     z_iter.append((j1, j2, s, sp))
@@ -131,7 +130,7 @@ def see_variables(vect: list, x_iter: list) -> dict:
 
 def see_non_zero_variables(vect: list, x_iter: list) -> dict:
     return_dict = {}
-    for i in range(len(x_iter)):
+    for i, _ in enumerate(x_iter):
         if vect[i] != 0:
             return_dict[x_iter[i]] = vect[i]
 
@@ -181,10 +180,9 @@ def qubo_to_matrix(qubo: dict, lp: LinearProg) -> np.ndarray:
 def check_solution_list(sol: list, lp: LinearProg):
     data = sorted(list(lp.bqm.variables))
     sol_dict = {data[i]: sol[i] for i in range(len(sol))}
-    qubo = lp.qubo[0]
     offset = lp.qubo[1]
     print(offset)
-    #matrix = qubo_to_matrix(qubo, lp)
+    #matrix = qubo_to_matrix(lp.qubo[0], lp)
     energy = compute_energy(sol, lp)
     sampleset = dimod.SampleSet.from_samples(dimod.as_samples(sol_dict), 'BINARY', energy)
     sampleset = lp.interpreter(sampleset)

Makefile

@@ -4,7 +4,7 @@ test:
 	@rm -rf tests/__pycache__/
 	@PYTHONPATH=. pytest -q --durations=10 --cov=. --cov-report term --cov-fail-under 75 tests/
 lint:
-	@pylint --fail-under=4.5 *.py
+	@pylint --fail-under=8.0 *.py
 
 clean:
 	@rm -rf *.jpg

examples/example_large.py

@@ -19,9 +19,9 @@
               ("s4", "s5"): 4, ("s5", "s4"): 4,
               ("s5", "s6"): 4, ("s6", "s5"): 4}
 
-agv_routes = {0: ("s0", "s1", "s2", "s3"), 
+agv_routes = {0: ("s0", "s1", "s2", "s3"),
               1: ("s0", "s1", "s2", "s3"),
-              2: ("s0", "s1", "s2", "s3"), 
+              2: ("s0", "s1", "s2", "s3"),
               3: ("s0", "s1", "s2", "s3"),
               4: ("s0", "s1", "s2"),
               5: ("s0", "s1", "s2"),
@@ -46,13 +46,13 @@
 d_max = {i: 40 for i in J}
 tau_pass = {(j, s, sp): tracks_len[(s, sp)] for j in J for s, sp in agv_routes_as_e[j]}
 tau_headway = {(j, jp, s, sp): 2 if (s, sp) != ("s2", "s3") and (s, sp) != ("s3", "s2") else 0
-               for (j, jp) in all_same_way.keys() for (s, sp) in all_same_way[(j, jp)]}
+               for (j, jp) in all_same_way for (s, sp) in all_same_way[(j, jp)]}
 
 tau_operation = {(agv, station): 2 for agv in J for station in stations}
 
 initial_conditions = {("in", 0, "s0"): 0, ("in", 1, "s0"): 1, ("in", 2, "s0"): 2, ("in", 3, "s0"): 3,
                       ("in", 4, "s0"): 4, ("in", 5, "s0"): 5, ("in", 6, "s4"): 0, ("in", 7, "s4"): 1,
-                      ("in", 8, "s4"): 2, ("in", 9, "s4"): 3, ("in", 10, "s4"): 8, 
+                      ("in", 8, "s4"): 2, ("in", 9, "s4"): 3, ("in", 10, "s4"): 8,
                       ("in", 11, "s4"): 5
                     }
 

examples/example_largest.py

@@ -19,9 +19,9 @@
               ("s4", "s5"): 4, ("s5", "s4"): 4,
               ("s5", "s6"): 4, ("s6", "s5"): 4}
 
-agv_routes = {0: ("s0", "s1", "s2", "s3"), 
+agv_routes = {0: ("s0", "s1", "s2", "s3"),
               1: ("s0", "s1", "s2", "s3"),
-              2: ("s0", "s1", "s2", "s3"), 
+              2: ("s0", "s1", "s2", "s3"),
               3: ("s0", "s1", "s2", "s3"),
               4: ("s0", "s1", "s2"),
               5: ("s0", "s1", "s2"),
@@ -45,14 +45,14 @@
 d_max = {i: 40 for i in J}
 tau_pass = {(j, s, sp): tracks_len[(s, sp)] for j in J for s, sp in agv_routes_as_e[j]}
 tau_headway = {(j, jp, s, sp): 2 if (s, sp) != ("s2", "s3") and (s, sp) != ("s3", "s2") else 0
-               for (j, jp) in all_same_way.keys() for (s, sp) in all_same_way[(j, jp)]}
+               for (j, jp) in all_same_way for (s, sp) in all_same_way[(j, jp)]}
 
 tau_operation = {(agv, station): 2 for agv in J for station in stations}
 
 initial_conditions = {("in", 0, "s0"): 0, ("in", 1, "s0"): 1, ("in", 2, "s0"): 2, ("in", 3, "s0"): 3,
                       ("in", 4, "s0"): 4, ("in", 5, "s0"): 5, ("in", 6, "s4"): 0, ("in", 7, "s4"): 1,
-                      ("in", 8, "s4"): 2, ("in", 9, "s4"): 3, ("in", 10, "s4"): 8, 
-                      ("in", 11, "s4"): 5, 
+                      ("in", 8, "s4"): 2, ("in", 9, "s4"): 3, ("in", 10, "s4"): 8,
+                      ("in", 11, "s4"): 5,
                       ("in", 12, "s2"): 7, ("in", 13, "s6"): 9, ("in", 14, "s5"): 9
                     }
 

examples/example_medium.py

@@ -37,12 +37,11 @@
 d_max = {i: 40 for i in J}
 tau_pass = {(j, s, sp): tracks_len[(s, sp)] for j in J for s, sp in agv_routes_as_e[j]}
 tau_headway = {(j, jp, s, sp): 2 if (s, sp) != ("s2", "s3") and (s, sp) != ("s3", "s2") else 0
-               for (j, jp) in all_same_way.keys() for (s, sp) in all_same_way[(j, jp)]}
+               for (j, jp) in all_same_way for (s, sp) in all_same_way[(j, jp)]}
 
 tau_operation = {(agv, station): 2 for agv in J for station in stations}
 
 initial_conditions = {("in", 0, "s0"): 0, ("in", 1, "s0"): 0, ("in", 2, "s4"): 8, ("in", 3, "s4"): 9,
                       ("in", 4, "s2"): 15, ("in", 5, "s6"): 0, ("in", 6, "s5"): 0}
 
 weights = {j: 1 for j in J}
-

examples/example_medium_small.py

@@ -36,11 +36,11 @@
 d_max = {i: 40 for i in J}
 tau_pass = {(j, s, sp): tracks_len[(s, sp)] for j in J for s, sp in agv_routes_as_e[j]}
 tau_headway = {(j, jp, s, sp): 2 if (s, sp) != ("s2", "s3") and (s, sp) != ("s3", "s2") else 0
-               for (j, jp) in all_same_way.keys() for (s, sp) in all_same_way[(j, jp)]}
+               for (j, jp) in all_same_way for (s, sp) in all_same_way[(j, jp)]}
 
 tau_operation = {(agv, station): 2 for agv in J for station in stations}
 
-initial_conditions = {("in", 0, "s0"): 0, ("in", 1, "s0"): 0, ("in", 2, "s4"): 8, 
+initial_conditions = {("in", 0, "s0"): 0, ("in", 1, "s0"): 0, ("in", 2, "s4"): 8,
                       ("in", 3, "s2"): 15, ("in", 4, "s6"): 0, ("in", 5, "s5"): 0}
 
 weights = {j: 1 for j in J}

examples/example_small.py

@@ -32,12 +32,10 @@
 d_max = {i: 10 for i in J}
 tau_pass = {(j, s, sp): tracks_len[(s, sp)] for j in J for s, sp in agv_routes_as_e[j]}
 tau_headway = {(j, jp, s, sp): 2 if (s, sp) != ("s2", "s3") and (s, sp) != ("s3", "s2") else 0
-               for (j, jp) in all_same_way.keys() for (s, sp) in all_same_way[(j, jp)]}
+               for (j, jp) in all_same_way for (s, sp) in all_same_way[(j, jp)]}
 
 tau_operation = {(agv, station): 2 for agv in J for station in stations}
 
 initial_conditions = {("in", 0, "s0"): 0, ("in", 1, "s1"): 8, ("in", 2, "s4"): 8, ("in", 4, "s2"): 15}
 
 weights = {j: 1 for j in J}
-
-

examples/example_smallest.py

@@ -30,12 +30,11 @@
 d_max = {i: 10 for i in J}
 tau_pass = {(j, s, sp): tracks_len[(s, sp)] for j in J for s, sp in agv_routes_as_e[j]}
 tau_headway = {(j, jp, s, sp): 2 if (s, sp) != ("s2", "s3") and (s, sp) != ("s3", "s2") else 0
-               for (j, jp) in all_same_way.keys() for (s, sp) in all_same_way[(j, jp)]}
+               for (j, jp) in all_same_way for (s, sp) in all_same_way[(j, jp)]}
 
 tau_operation = {(agv, station): 2 for agv in J for station in stations}
 
 initial_conditions = {("in", 0, "s1"): 8, ("in", 1, "s0"): 0
                      }
 
 weights = {j: 1 for j in J}
-

run_examples.py

@@ -3,15 +3,15 @@
 import pickle
 import time
 import os
+import argparse
 
 from AGV_quantum import plot_train_diagram
 from AGV_quantum import print_ILP_size, LinearAGV
 from AGV_quantum import QuadraticAGV
 from AGV_quantum import constrained_solver, hybrid_anneal
 
 
-import argparse
-parser = argparse.ArgumentParser("Solve linear or quadratic") 
+parser = argparse.ArgumentParser("Solve linear or quadratic")
 parser.add_argument(
     "--solve_linear",
     type=int,
@@ -92,18 +92,18 @@
         hybrid = args.hyb_solver
         p = 5 # penalty for QUBO creation
         model = QuadraticAGV(AGV)
-        
+
         # saves model for checks
         lp_file = os.path.join(cwd, f"lp_files/lp_{args.example}.pkl")
         if not os.path.isfile(lp_file):
             with open(lp_file, "wb") as f:
-                    pickle.dump(model, f)
+                pickle.dump(model, f)
 
         model.to_bqm_qubo_ising(p)
 
         # check if results are saved
         is_file = os.path.isfile(os.path.join(save_path, f"new_{hybrid}_info.pkl"))
-        if is_file: 
+        if is_file:
             print(".......... files exist ............")
             if hybrid == "bqm":
                 print("n.o. qubits", model._count_qubits())
@@ -123,9 +123,9 @@
             info = sampleset.info
             print(sampleset)
             print(info)
-            
+
             with open(os.path.join(save_path, f"new_{hybrid}_info.pkl"), "wb") as f:
                 pickle.dump(info, f)
 
             with open(os.path.join(save_path, f"new_{hybrid}.pkl"), "wb") as f:
-                pickle.dump(sampleset.to_serializable(), f)
+                pickle.dump(sampleset.to_serializable(), f)