Skip to content

Commit 16f28a6

Browse files
committed
resolved some mypy issues
1 parent 9c2884f commit 16f28a6

File tree

4 files changed

+92
-49
lines changed

4 files changed

+92
-49
lines changed

pyproject.toml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,7 @@ exclude = [
135135

136136
[[tool.mypy.overrides]]
137137
module = ["qiskit.*", "qecsim.*", "qiskit_aer.*", "matplotlib.*", "scipy.*", "ldpc.*", "pytest_console_scripts.*",
138-
"z3.*", "bposd.*", "numba.*", "pymatching.*", "stim.*", "multiprocess.*", "qsample.*", "pandas.*", "networkx.*"]
138+
"z3.*", "bposd.*", "numba.*", "pymatching.*", "stim.*", "multiprocess.*", "qsample.*", "pandas.*", "networkx.*", "tqdm.*"]
139139
ignore_missing_imports = true
140140

141141

src/mqt/qecc/co3/utils/hill_climber.py

Lines changed: 75 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import pickle # noqa: S403
88
import random
99
from pathlib import Path
10+
from typing import Any, TypedDict, cast
1011

1112
import matplotlib.pyplot as plt
1213
import networkx as nx
@@ -24,8 +25,14 @@
2425

2526
random.seed(45)
2627

28+
class HistoryTemp(TypedDict, total=False):
29+
"""Type for history dictionaries."""
30+
scores: list[int]
31+
layout_init: dict[int | str, tuple[int, int] | list[tuple[int, int]]]
32+
layout_final: dict[int | str, tuple[int, int] | list[tuple[int, int]]]
2733

28-
def save_to_file(path: str, data: dict) -> None:
34+
35+
def save_to_file(path: str, data: Any) -> None: # noqa: ANN401
2936
"""Safely saves data to a file."""
3037
with Path(path).open("wb") as pickle_file:
3138
pickle.dump(data, pickle_file)
@@ -36,8 +43,8 @@ class HillClimbing:
3643

3744
def __init__(
3845
self,
39-
max_restarts: int | None,
40-
max_iterations: int | None,
46+
max_restarts: int,
47+
max_iterations: int,
4148
circuit: list[tuple[int, int] | int],
4249
layout_type: str,
4350
m: int,
@@ -48,7 +55,7 @@ def __init__(
4855
free_rows: list[str] | None = None,
4956
t: int | None = None,
5057
optimize_factories: bool = False,
51-
custom_layout: list[list, nx.Graph] | None = None,
58+
custom_layout: list[list[tuple[int,int]] | nx.Graph] | None = None,
5259
routing: str = "static",
5360
) -> None:
5461
"""Initializes the Hill Climbing with Random Restarts algorithm.
@@ -70,7 +77,7 @@ def __init__(
7077
free_rows (list[str] | None): Adds one or more rows to lattice, either top or right (easier to implement than also adding bottom, left). Defaults to None.
7178
t (int): waiting time for factories. Defaults to None
7279
optimize_factories (int): decides whether factories are optimized or not. Defaults to false.
73-
custom_layout (list[list, nx.Graph] | None): Defaults to None because custom layouts not assumed to be standard. The first list in the list should be
80+
custom_layout (list[list[tuple[int,int]] | nx.Graph] | None): Defaults to None because custom layouts not assumed to be standard. The first list in the list should be
7481
a `data_qubits_loc` of the node locations of data qubits and nx.Graph the corresponding graph (possibly differing from the standard networkx hex graph shape)
7582
With custom_layout one can avoid using the `free_rows` related stuff.
7683
routing (str): Defaults to static. Can be "static" or "dynamic". Chooses the routing scheme, whether the layout-agnostic initial layers are dynamically adapted or not.
@@ -90,9 +97,10 @@ def __init__(
9097
)
9198
assert num_factories is not None, "If T gates included in circuit, `num_factories` must NOT be None."
9299
assert t is not None, "If T gates included in circuit, `num_factories` must NOT be None."
93-
assert len(possible_factory_positions) >= num_factories, (
94-
f"`possible_factory_positions` must have more or equal elements than `num_factories`. But {len(self.possible_factory_positions)} ? {num_factories}"
95-
)
100+
if possible_factory_positions is not None:
101+
assert len(possible_factory_positions) >= num_factories, (
102+
f"`possible_factory_positions` must have more or equal elements than `num_factories`. But {len(possible_factory_positions)} ? {num_factories}"
103+
)
96104
else:
97105
assert optimize_factories is False, "If no T gates present, optimize_factories must be false."
98106
self.possible_factory_positions = possible_factory_positions
@@ -125,8 +133,9 @@ def __init__(
125133
elif layout_type == "hex":
126134
data_qubit_locs = lat.gen_layout_hex()
127135
elif layout_type == "custom":
128-
data_qubit_locs = custom_layout[0]
129-
self.lat.G = custom_layout[1]
136+
if custom_layout is not None: #only for mypy
137+
data_qubit_locs = custom_layout[0]
138+
self.lat.G = custom_layout[1]
130139
else:
131140
msg = "unknown layout type"
132141
raise ValueError(msg)
@@ -139,7 +148,7 @@ def __init__(
139148
num for tup in self.circuit for num in (tup if isinstance(tup, tuple) else (tup,))
140149
]
141150
else:
142-
flattened_qubit_labels = [num for tup in self.circuit for num in tup]
151+
flattened_qubit_labels = [num for tup in self.circuit if isinstance(tup, tuple) for num in tup] #isinstance only added for mypy
143152
self.q = max(flattened_qubit_labels) + 1
144153
if self.q < len(self.data_qubit_locs):
145154
self.data_qubit_locs = self.data_qubit_locs[: self.q] # cut-off unnecessary qubit spots.
@@ -225,45 +234,62 @@ def add_left_g(g: nx.Graph) -> nx.Graph:
225234

226235
return g
227236

228-
def evaluate_solution(self, layout: dict) -> int:
237+
def evaluate_solution(self, layout: dict[int | str, tuple[int, int] | list[tuple[int,int]]]) -> int:
229238
"""Evaluates the layout=solution according to self.metric."""
230239
terminal_pairs = translate_layout_circuit(self.circuit, layout)
240+
factory_positions: list[tuple[int,int]]
241+
#if type(layout["factory_positions"]) is list[tuple[int,int]] or list:
231242
factory_positions = layout["factory_positions"]
243+
#else:
244+
# msg = f"factory positions of layout must be list[tuple[int,int]]. But you got {type(layout['factory_positions'])}"
245+
# raise TypeError(msg)
246+
router : ShortestFirstRouter | ShortestFirstRouterTGatesDyn | ShortestFirstRouterTGates
232247
if any(type(el) is int for el in self.circuit):
233-
if self.routing == "static":
234-
router = ShortestFirstRouterTGates(
235-
m=self.m, n=self.n, terminal_pairs=terminal_pairs, factory_positions=factory_positions, t=self.t
236-
)
237-
elif self.routing == "dynamic":
238-
router = ShortestFirstRouterTGatesDyn(
239-
m=self.m, n=self.n, terminal_pairs=terminal_pairs, factory_positions=factory_positions, t=self.t
240-
)
241-
if self.layout_type == "custom": # Must update the router's g to the customized g
242-
router.G = self.lat.G.copy()
243-
else: # if not custom, update self.lat.G by the router's G because m,n might differ from initial values.
244-
self.lat.G = router.G # also add to self
245-
if "left" in self.free_rows:
246-
router.G = self.add_left_g(router.G)
248+
if self.t is not None:
249+
if self.routing == "static":
250+
router = ShortestFirstRouterTGates(
251+
m=self.m, n=self.n, terminal_pairs=terminal_pairs, factory_positions=factory_positions, t=self.t
252+
)
253+
elif self.routing == "dynamic":
254+
router = ShortestFirstRouterTGatesDyn(
255+
m=self.m, n=self.n, terminal_pairs=terminal_pairs, factory_positions=factory_positions, t=self.t
256+
)
257+
if self.layout_type == "custom": # Must update the router's g to the customized g
258+
router.G = self.lat.G.copy()
259+
else: # if not custom, update self.lat.G by the router's G because m,n might differ from initial values.
247260
self.lat.G = router.G # also add to self
261+
if self.free_rows is not None: #for mypy # noqa: SIM102
262+
if "left" in self.free_rows:
263+
router.G = self.add_left_g(router.G)
264+
self.lat.G = router.G # also add to self
265+
else:
266+
msg = "t must be not None if the circuit contains single integers for T gates."
267+
raise ValueError(msg)
248268
else: # only CNOTs
249269
if self.routing == "static":
250-
router = ShortestFirstRouter(m=self.m, n=self.n, terminal_pairs=terminal_pairs)
270+
terminal_pairs_cast = cast("list[tuple[tuple[int, int], tuple[int, int]]]", terminal_pairs)
271+
router = ShortestFirstRouter(m=self.m, n=self.n, terminal_pairs=terminal_pairs_cast)
251272
elif self.routing == "dynamic": # !todo adapt this, because if only cnots, t might not be defined
273+
t = cast("int", self.t)
252274
router = ShortestFirstRouterTGatesDyn(
253-
m=self.m, n=self.n, terminal_pairs=terminal_pairs, factory_positions=factory_positions, t=self.t
275+
m=self.m, n=self.n, terminal_pairs=terminal_pairs, factory_positions=factory_positions, t=t
254276
)
255277
if self.layout_type == "custom": # Must update the router's g to the customized g
256278
router.G = self.lat.G.copy()
257279
else:
258280
self.lat.G = router.G # also add to self (but should be redundant right? self.lat.G and router.G should be the same anyways if no T gates present)
281+
cost: int
259282
if self.metric == "crossing":
260283
if self.optimize_factories and any(type(el) is int for el in self.circuit):
284+
router = cast("ShortestFirstRouterTGates | ShortestFirstRouterTGatesDyn", router)
261285
cost = np.sum(router.count_crossings_per_layer(t_crossings=True))
262286
elif self.optimize_factories is False and any(type(el) is int for el in self.circuit):
287+
router = cast("ShortestFirstRouterTGates | ShortestFirstRouterTGatesDyn", router)
263288
cost = np.sum(router.count_crossings_per_layer(t_crossings=False))
264289
else:
265290
cost = np.sum(router.count_crossings_per_layer())
266291
elif self.metric == "distance":
292+
router = cast("ShortestFirstRouter", router)
267293
distances = router.measure_terminal_pair_distances()
268294
cost = np.sum(distances)
269295
if any(type(el) is int for el in self.circuit):
@@ -272,13 +298,14 @@ def evaluate_solution(self, layout: dict) -> int:
272298
if self.routing == "static":
273299
vdp_layers = router.find_total_vdp_layers()
274300
elif self.routing == "dynamic":
301+
router = cast("ShortestFirstRouterTGatesDyn", router)
275302
vdp_layers = router.find_total_vdp_layers_dyn()
276303
cost = len(vdp_layers)
277304
return cost
278305

279-
def gen_random_qubit_assignment(self) -> dict[int | str, tuple[int, int] | list[int]]:
306+
def gen_random_qubit_assignment(self) -> dict[int | str, tuple[int, int] | list[tuple[int,int]]]:
280307
"""Yields a random qubit assignment given the `data_qubit_locs`."""
281-
layout = {}
308+
layout: dict[int | str, tuple[int, int] | list[tuple[int,int]]] = {}
282309
perm = list(range(self.q))
283310
random.shuffle(perm)
284311
for i, j in zip(
@@ -289,12 +316,14 @@ def gen_random_qubit_assignment(self) -> dict[int | str, tuple[int, int] | list[
289316
# Add generation of random choice of factory positions
290317
factory_positions = []
291318
if any(type(el) is int for el in self.circuit):
292-
factory_positions = random.sample(self.possible_factory_positions, self.num_factories)
319+
possible_factory_positions = cast("list[tuple[int,int]]", self.possible_factory_positions)
320+
num_factories = cast("int", self.num_factories)
321+
factory_positions = random.sample(possible_factory_positions, num_factories)
293322
layout.update({"factory_positions": factory_positions})
294323

295324
return layout
296325

297-
def gen_neighborhood(self, layout: dict) -> list[dict]:
326+
def gen_neighborhood(self, layout: dict[int | str, tuple[int, int] | list[tuple[int,int]]]) -> list[dict[int | str, tuple[int, int] | list[tuple[int,int]]]]:
298327
"""Creates the Neighborhood of a given layout by going through each terminal pair and swapping their positions.
299328
300329
If there are no T gates, there will be l=len(terminal_pairs) elements in the neighborhood.
@@ -335,7 +364,12 @@ def gen_neighborhood(self, layout: dict) -> list[dict]:
335364

336365
return neighborhood
337366

338-
def _parallel_hill_climbing(self, restart: int) -> tuple:
367+
def _parallel_hill_climbing(self, restart: int) -> tuple[
368+
int,
369+
dict[int | str, tuple[int, int] | list[tuple[int,int]]],
370+
int,
371+
HistoryTemp
372+
]:
339373
"""Helper method for parallel execution of hill climbing restarts.
340374
341375
Args:
@@ -350,7 +384,7 @@ def _parallel_hill_climbing(self, restart: int) -> tuple:
350384

351385
current_solution = self.gen_random_qubit_assignment()
352386
current_score = self.evaluate_solution(current_solution)
353-
history_temp = {"scores": [], "layout_init": current_solution.copy()}
387+
history_temp: HistoryTemp = {"scores": [], "layout_init": current_solution.copy()}
354388

355389
for _ in range(self.max_iterations):
356390
neighbors = self.gen_neighborhood(current_solution)
@@ -374,7 +408,7 @@ def _parallel_hill_climbing(self, restart: int) -> tuple:
374408
history_temp.update({"layout_final": current_solution.copy()})
375409
return restart, current_solution, current_score, history_temp
376410

377-
def run(self, prefix: str, suffix: str, parallel: bool, processes: int = 8) -> tuple[dict, int, int, dict]:
411+
def run(self, prefix: str, suffix: str, parallel: bool, processes: int = 8) -> tuple[dict[int | str, tuple[int, int] | list[tuple[int, int]]], int, int, dict[int,HistoryTemp]]:
378412
"""Executes the Hill Climbing algorithm with random restarts.
379413
380414
Args:
@@ -390,7 +424,7 @@ def run(self, prefix: str, suffix: str, parallel: bool, processes: int = 8) -> t
390424
best_solution = None
391425
best_rep = None
392426
best_score = float("inf") # Use '-inf' for maximization, 'inf' for minimization
393-
score_history = {}
427+
score_history: dict[int, HistoryTemp] = {}
394428
path = (
395429
prefix
396430
+ f"hill_climbing_data_q{self.q}_numcnots{len(self.circuit)}_layout{self.layout_type}_metric{self.metric}_parallel{parallel}"
@@ -425,7 +459,7 @@ def run(self, prefix: str, suffix: str, parallel: bool, processes: int = 8) -> t
425459

426460
current_solution = self.gen_random_qubit_assignment()
427461
current_score = self.evaluate_solution(current_solution)
428-
history_temp = {"scores": [], "layout_init": current_solution.copy()}
462+
history_temp: HistoryTemp = {"scores": [], "layout_init": current_solution.copy()}
429463
for _ in range(self.max_iterations):
430464
neighbors = self.gen_neighborhood(current_solution)
431465
if not neighbors:
@@ -455,10 +489,13 @@ def run(self, prefix: str, suffix: str, parallel: bool, processes: int = 8) -> t
455489
best_solution, best_score = current_solution, current_score
456490
best_rep = restart
457491

492+
best_solution = cast("dict[int | str, tuple[int, int] | list[tuple[int, int]]]", best_solution)
493+
best_score = cast("int", best_score)
494+
best_rep = cast("int", best_rep)
458495
return best_solution, best_score, best_rep, score_history
459496

460497
def plot_history(
461-
self, score_history: dict, filename: str = "./hc_history_plot.pdf", size: tuple[float, float] = (5, 5)
498+
self, score_history: HistoryTemp, filename: str = "./hc_history_plot.pdf", size: tuple[float, float] = (5, 5)
462499
) -> None:
463500
"""Plots the scores for each restart and iteration.
464501

src/mqt/qecc/co3/utils/misc.py

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -185,7 +185,7 @@ def generate_random_circuit(
185185

186186

187187
def translate_layout_circuit(
188-
pairs: list[tuple[int, int] | int], layout: dict[int | str, tuple[int, int] | list[int]]
188+
pairs: list[tuple[int, int] | int], layout: dict[int | str, tuple[int, int] | list[tuple[int,int]]]
189189
) -> list[tuple[tuple[int, int], tuple[int, int]] | tuple[int, int]]:
190190
"""Translates a `pairs` circuit (with int labels) into the lattice's labels for a given layout.
191191
@@ -197,21 +197,27 @@ def translate_layout_circuit(
197197
terminal_pairs: list[tuple[tuple[int, int], tuple[int, int]] | tuple[int, int]] = []
198198
for pair in pairs:
199199
if isinstance(pair, tuple):
200-
pos1 = layout[pair[0]]
201-
pos2 = layout[pair[1]]
202-
pos1 = (int(pos1[0]), int(pos1[1]))
203-
pos2 = (int(pos2[0]), int(pos2[1]))
200+
pos1_raw = layout[pair[0]]
201+
pos2_raw = layout[pair[1]]
202+
if not isinstance(pos1_raw, tuple) or not isinstance(pos2_raw, tuple):
203+
msg = "Expected tuple[int, int] in layout mapping."
204+
raise TypeError(msg)
205+
pos1 = (int(pos1_raw[0]), int(pos1_raw[1]))
206+
pos2 = (int(pos2_raw[0]), int(pos2_raw[1]))
204207
terminal_pairs.append((pos1, pos2))
205208
else:
206-
pos = layout[pair]
207-
pos = (int(pos[0]), int(pos[1]))
209+
pos_raw = layout[pair]
210+
if not isinstance(pos_raw, tuple):
211+
msg = "Expected tuple[int, int] in layout mapping."
212+
raise TypeError(msg)
213+
pos = (int(pos_raw[0]), int(pos_raw[1]))
208214
terminal_pairs.append(pos)
209215

210216
return terminal_pairs
211217

212218

213219
def compare_original_dynamic_gate_order(
214-
q: int, layout: dict[int | str, tuple[int, int] | list[int]], router: co.ShortestFirstRouterTGatesDyn
220+
q: int, layout: dict[int | str, tuple[int, int] | list[tuple[int,int]]], router: co.ShortestFirstRouterTGatesDyn
215221
) -> bool:
216222
"""Generates a qiskit circuit for both the order after doing dynamic routing and the original order.
217223

test/python/co3/test_lattice_router.py

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -210,7 +210,7 @@ def test_dynamic_router():
210210
m = 3
211211
n = 4
212212
factory_locs = [(1, 7), (3, 7)]
213-
layout: dict[int | str, tuple[int, int] | list[int]] = {
213+
layout: dict[int | str, tuple[int, int] | list[tuple[int,int]]] = {
214214
2: (1, 2),
215215
5: (1, 3),
216216
1: (2, 2),
@@ -256,7 +256,7 @@ def test_ordering_dyn_routing():
256256
# generate random circuit
257257
pairs = co.generate_random_circuit(q, min_depth=q, tgate=True, ratio=0.8)
258258
# generate random layout
259-
layout: dict[int | str, tuple[int, int] | list[int]] = {}
259+
layout: dict[int | str, tuple[int, int] | list[tuple[int,int]]] = {}
260260
perm = list(range(len(data_qubit_locs)))
261261
random.shuffle(perm)
262262
for i, j in zip(

0 commit comments

Comments
 (0)