2 from squander.partitioning.tools
import get_qubits, build_dependency, parts_to_float_ops, total_float_ops
4 _GUROBI_FALLBACK_WARNING_PRINTED =
False 5 _GUROBI_AVAILABLE =
None 6 _GUROBI_AVAILABILITY_ERROR =
None 10 global _GUROBI_FALLBACK_WARNING_PRINTED
11 if _GUROBI_FALLBACK_WARNING_PRINTED:
13 _GUROBI_FALLBACK_WARNING_PRINTED =
True 15 "Warning: Gurobi ILP solver is unavailable; falling back to CBC. " 16 "This can be significantly slower. For best performance, install a " 17 "valid Gurobi license in your home directory or under /opt/gurobi/ " 18 "as described in the Gurobi documentation. " 19 f
"Original Gurobi error: {exc}" 24 global _GUROBI_AVAILABLE, _GUROBI_AVAILABILITY_ERROR
25 if _GUROBI_AVAILABLE
is False:
26 raise _GUROBI_AVAILABILITY_ERROR
27 if _GUROBI_AVAILABLE
is True:
31 env = gp.Env(params={
"OutputFlag": 0})
33 _GUROBI_AVAILABLE =
True 34 except Exception
as exc:
35 _GUROBI_AVAILABLE =
False 36 _GUROBI_AVAILABILITY_ERROR = exc
43 solver = pulp.GUROBI(manageEnv=
True, msg=
False, **gurobi_kwargs)
47 prob.solve(solver, callback=callback)
49 except Exception
as exc:
51 prob.solve(pulp.PULP_CBC_CMD(msg=
False))
57 Topologically sort partition groups and re-partition with Kahnâs algorithm. 60 c: SQUANDER circuit object. 61 max_qubits_per_partition (int): Maximum number of qubits allowed in a partition. 62 parts (list[Iterable[int]]): Precomputed gate groups (each is an iterable of gate indices). 65 tuple: (partitioned_circ, param_order, parts) 66 partitioned_circ: 2-level partitioned circuit (SQUANDER object). 67 param_order (list[tuple[int,int,int]]): Tuples (source_idx, dest_idx, param_count). 68 parts (list[frozenset[int]]): Final partition assignments (topologically sorted). 70 gatedict = {i: gate
for i, gate
in enumerate(c.get_Gates())}
71 gate_to_qubit = {i:
get_qubits(g)
for i, g
in gatedict.items()}
72 partdict = {gate: i
for i, part
in enumerate(parts)
for gate
in part}
73 g, rg = {i: set()
for i
in range(len(parts))}, {i: set()
for i
in range(len(parts))}
75 for child
in c.get_Children(gatedict[gate]):
76 if partdict[gate] == partdict[child]:
continue 77 g[partdict[gate]].add(partdict[child])
78 rg[partdict[child]].add(partdict[gate])
81 i: tuple(sorted(set.union(*(gate_to_qubit[g]
for g
in part))))
82 for i, part
in enumerate(parts)
84 L, S = [], [(part_qubits[m], m)
for m
in rg
if len(rg[m]) == 0]
87 n = heapq.heappop(S)[1]
89 assert len(rg[n]) == 0
94 heapq.heappush(S, (part_qubits[m], m))
95 assert len(L) == len(parts), (len(L), len(parts), g, rg, partdict)
107 Partitions a circuit using ILP to maximize gates per partition 109 c: SQUANDER Circuit to partition 110 max_qubits_per_partition: Max qubits per partition 113 Partitioned circuit, parameter order (source_idx, dest_idx, param_count), partition assignments 115 gatedict = {i: gate
for i, gate
in enumerate(c.get_Gates())}
116 num_gates = len(gatedict)
118 while sum(len(x)
for x
in parts) != num_gates:
121 from squander.partitioning.kahn
import kahn_partition_preparts
127 Find the next largest feasible gate partition via ILP (single partition step). 130 c: SQUANDER circuit to be partitioned. 131 max_qubits_per_partition (int): Maximum number of qubits allowed in the partition. 132 prevparts (Iterable[Iterable[int]] | None): Previously selected partitions whose 133 gates must be excluded in this step. 136 set[int]: Set of gate indices forming the next largest partition. 139 gatedict = {i: gate
for i, gate
in enumerate(c.get_Gates())}
140 all_qubits = set(range(c.get_Qbit_Num()))
142 num_gates = len(gatedict)
143 prob = pulp.LpProblem(
"MaxSinglePartition", pulp.LpMinimize)
144 a = pulp.LpVariable.dicts(
"a", (i
for i
in range(num_gates)), cat=
"Binary")
145 x = pulp.LpVariable.dicts(
"x", (i
for i
in range(num_gates)), cat=
"Binary")
146 z = pulp.LpVariable.dicts(
"z", (q
for q
in all_qubits), cat=
"Binary")
149 prob += x[i] + a[i] <= 1
153 for q
in gate_qubits:
156 prob += pulp.lpSum(z[q]
for q
in all_qubits) <= max_qubits_per_partition
158 for part
in prevparts:
161 prob += a[i] == a[next(iter(part))]
164 for i
in c.get_Children(gatedict[j]):
166 prob += x[j] + a[j] >= x[i]
167 prob.setObjective(-pulp.lpSum(x[i]
for i
in range(num_gates)))
172 gates = {i
for i
in range(num_gates)
if int(pulp.value(x[i]))}
180 Compute SCCs and intra-SCC reachability using Nuutila's variant (iterative). 183 succ (dict[Hashable, Iterable[Hashable]]): Successor adjacency list. 184 subg (set[Hashable] | None): Optional node subset to restrict the search. 188 sccs (dict[Hashable, set]): Map from SCC root to set of vertices in that SCC. 189 reach (dict[Hashable, set]): For each SCC root, the set of vertices reachable 190 from that SCC (including the SCC itself iff it has a self-loop). 192 index, s, sc, sccs, reach = 0, [], [], {}, {}
193 indexes, lowlink, croot, stackheight = {}, {}, {}, {}
194 def nuutila(v, index):
196 stack = [(v,
None, iter(succ[v]))]
197 while len(stack) != 0:
198 v, w, succv = stack.pop()
200 indexes[v], lowlink[v] = index, v
201 stackheight[v] = len(sc)
205 if indexes[lowlink[w]] < indexes[lowlink[v]]: lowlink[v] = lowlink[w]
206 else: sc.append(croot[w])
208 if not subg
is None and not w
in subg:
continue 210 if not w
in indexes: stack.append((v, w, succv)); stack.append((w,
None, iter(succ[w])));
break 211 else: forward_edge = indexes[v] < indexes[w]
213 if indexes[lowlink[w]] < indexes[lowlink[v]]: lowlink[v] = lowlink[w]
214 elif not forward_edge:
219 is_self_loop = s[-1] != v
or v
in succ[v]
224 reach[v] = set(sccs[v])
if is_self_loop
else set()
225 if not subg
is None and len(subg) == len(sccs[v]):
return index
226 for x
in sccs[v]: croot[x] = v
228 while len(sc) != stackheight[v]:
231 for x
in sorted(l, reverse=
True, key=
lambda y: indexes[y]):
232 if not x
in reach[v]:
233 reach[v] |= reach[x]; reach[v] |= sccs[x]
236 if (subg
is None or v
in subg)
and not v
in indexes:
237 index = nuutila(v, index)
242 Get a topological order of a DAG given forward and reverse adjacency. 245 g (dict[int, set[int]]): Forward adjacency (u -> successors). 246 rg (dict[int, set[int]]): Reverse adjacency (v -> predecessors). 249 list[int]: A topological ordering of nodes. 251 g = { x: set(y)
for x, y
in g.items() }
252 rg = { x: set(y)
for x, y
in rg.items() }
253 S = [(tuple(sorted(gate_to_qubit[m])), m)
for m
in rg
if len(rg[m]) == 0]
257 n = heapq.heappop(S)[1]
264 heapq.heappush(S, (tuple(sorted(gate_to_qubit[m])), m))
269 Contract maximal chains of single-qubit gates to simplify the DAG. 272 go (dict[int, set[int]]): Forward gate DAG (gate -> children). 273 rgo (dict[int, set[int]]): Reverse gate DAG (gate -> parents). 274 gate_to_qubit (dict[int, set[int]]): Gate index -> acted-on qubits. 275 topo_order (list[int]): A topological order of the original DAG. 278 tuple: (g, rg, topo_order_reduced, chains) 279 g (dict[int, set[int]]): Reduced forward DAG after contraction. 280 rg (dict[int, set[int]]): Reduced reverse DAG after contraction. 281 topo_order_reduced (list[int]): Topological order without single-qubit gates. 282 chains (set[tuple[int,...]]): Set of contracted chains (each a tuple of gate ids). 285 g = { x: set(y)
for x, y
in go.items() }
286 rg = { x: set(y)
for x, y
in rgo.items() }
287 single_qubit_gates = {x
for x, y
in gate_to_qubit.items()
if len(y) == 1}
288 single_qubit_chains = {}
289 for gate
in topo_order:
290 if gate
in single_qubit_gates:
293 v = next(iter(rgo[gate]))
294 single_qubit_chains[gate] = single_qubit_chains[v]
if v
in single_qubit_chains
else []
295 else: single_qubit_chains[gate] = []
296 single_qubit_chains[gate].append(gate)
298 v = next(iter(rg[gate]))
302 v = next(iter(g[gate]))
305 del g[gate]; del rg[gate]
306 topo_order = [x
for x
in topo_order
if not x
in single_qubit_gates]
307 return g, rg, topo_order, set(tuple(x)
for x
in single_qubit_chains.values())
311 Re-insert contracted single-qubit chains back into partition groups. 314 g (dict[int, set[int]]): Forward DAG of original gates. 315 rg (dict[int, set[int]]): Reverse DAG of original gates. 316 single_qubit_chains (set[tuple[int,...]]): Contracted chains to re-attach. 317 gate_to_tqubit (dict[int, int]): Gate -> target qubit id. 318 L (Iterable[Iterable[int]]): Current partition groups. 319 fusion_info (tuple[set[int], set[int]] | None): Optional pre/post placement 320 hints (inpre, inpost) for chain endpoints. 323 list[frozenset[int]]: Updated partition groups with single-qubit chains merged. 325 L = [set(x)
for x
in L]
326 gate_to_part = {x: part
for part
in L
for x
in part}
327 if fusion_info
is not None: inpre, inpost = fusion_info
328 for chain
in single_qubit_chains:
330 if rg[chain[0]]
and g[chain[-1]]:
331 v = next(iter(rg[chain[0]]))
332 w = next(iter(g[chain[-1]]))
333 if not surrounded_only
and (fusion_info
is None or chain[0]
in inpre)
or gate_to_part[v] == gate_to_part[w]:
334 gate_to_part[v] |= frozenset(chain)
335 elif not surrounded_only
and (fusion_info
is None or chain[-1]
in inpost):
336 gate_to_part[w] |= frozenset(chain)
337 else: L.append(frozenset(chain))
338 elif not surrounded_only
and rg[chain[0]]:
339 v = next(iter(rg[chain[0]]))
340 if fusion_info
is None or chain[0]
in inpre:
341 gate_to_part[v] |= frozenset(chain)
342 else: L.append(frozenset(chain))
343 elif not surrounded_only
and g[chain[-1]]:
344 v = next(iter(g[chain[-1]]))
345 if fusion_info
is None or chain[-1]
in inpost:
346 gate_to_part[v] |= frozenset(chain)
347 else: L.append(frozenset(chain))
349 L.append(frozenset(chain))
353 Strongly Connected Components (Tarjan) without recursion. 357 succ : list of iterables 358 succ[u] is an iterable of out-neighbors of u. Nodes are 0..n-1. 363 comp_id[u] = index of the SCC containing u (0..k-1). 364 comps : list[list[int]] 365 List of SCCs; each is a list of nodes. Order is discovery order. 369 - No recursion; uses an explicit DFS stack of (u, iterator) frames. 370 - Self-loops and parallel edges are handled. 371 - Time O(n + m), space O(n). 373 index = {u: -1
for u
in succ}
374 low = {u: 0
for u
in succ}
375 onstk = {u:
False for u
in succ}
378 comp_id = {u: -1
for u
in succ}
391 index[s] = low[s] = idx; idx += 1
392 tarjan_stack.append(s); onstk[s] =
True 393 dfs_stack.append((s, iter(succ[s])))
396 u, it = dfs_stack[-1]
400 if low[u] == index[u]:
404 w = tarjan_stack.pop()
406 comp_id[w] = len(comps)
421 index[v] = low[v] = idx; idx += 1
422 tarjan_stack.append(v); onstk[v] =
True 423 dfs_stack.append((v, iter(succ[v])))
426 if index[v] < low[u]:
429 return comp_id, comps
432 Build a partition-level interaction DAG and prune intra-overlap edges. 435 g (dict[int, set[int]]): Gate DAG (u -> successors v). 436 gate_to_parts (dict[int, Iterable[int]]): Gate -> indices of parts containing it. 439 dict[int, set[int]]: Pruned successor map among parts (DAG over part indices). 442 for v, idxs
in gate_to_parts.items():
444 overlaps.setdefault(i, set()).update(j
for j
in idxs
if j != i)
445 succ = {x: set()
for x
in overlaps}
452 if i == j
or j
in overlaps[i]:
continue 455 succ_pruned = {x: set()
for x
in succ}
458 if scc_id[i] == scc_id[j]: succ_pruned[i].add(j)
462 Enumerate chordless directed cycles consistent with a given successor map. 465 succ (dict[Hashable, set[Hashable]]): Successor adjacency among nodes. 466 max_len (int | None): Optional max cycle length bound; None disables bound. 469 list[list[Hashable]]: List of chordless cycles (each as node list in order). 474 def can_extend(path, in_path, x, s):
475 """Check if we can extend ...->u to x via (u->x) without creating any chord.""" 477 if x
in in_path
or x < s:
481 if v != u
and v
in in_path:
485 if v != s
and v
in succ[x]:
490 """Check if we can close with last->s and remain chordless.""" 501 if v != u
and s
in succ[v]:
504 pred = {v: set()
for v
in succ}
506 for v
in succ[u]: pred[v].add(u)
509 if not succ[s]
or not pred[s]:
continue 510 if max_len
is not None:
511 dist_to_s = {v:
None for v
in succ}
512 from collections
import deque
518 if dist_to_s[p]
is None:
519 dist_to_s[p] = dist_to_s[w] + 1
523 stack = [(s, iter(succ[s]),
False)]
525 u, nbrs, tried_close = stack[-1]
527 if not tried_close
and len(path) >= 2
and (s
in succ[u])
and can_close(path):
529 cycles.append(path.copy())
530 stack[-1] = (u, nbrs,
True)
542 stack[-1] = (u, nbrs, tried_close)
544 if max_len
is not None and (dist_to_s[x]
is None or len(path) + dist_to_s[x] > max_len):
547 if can_extend(path, in_path, x, s):
551 stack.append((x, iter(succ[x]),
False))
555 Detect pairwise 2-cycles between partitions induced by gate-level edges. 558 g (dict[int, set[int]]): Gate DAG (u -> successors v). 559 gate_to_parts (dict[int, Iterable[int]]): Gate -> indices of parts containing it. 560 allparts (list[frozenset[int]]): The parts (gate sets) by index. 563 list[tuple[int, int]]: List of (a, b) where a < b and a <-> b two-cycle is found, 564 excluding cases where parts overlap. 571 pu = gate_to_parts[u]
572 pv = gate_to_parts[v]
576 a, b = (i, j)
if i < j
else (j, i)
577 bit = 1
if i < j
else 2
578 prev = seen.get((a, b), 0)
580 if new == 3
and prev != 3
and len(allparts[a] & allparts[b]) == 0:
581 twocycles.append((a, b))
588 for gate
in allparts[i]: gate_to_part[gate] = i
589 G_part = {i: set()
for i
in L}
591 for u
in allparts[i]:
593 if not v
in gate_to_part:
continue 598 return {frozenset(v)
for v
in scc
if len(v) > 1}
600 def ilp_global_optimal(allparts, g, weighted_info=None, gurobi_direct=False, use_order=False, weights=None):
602 Select an optimal set of non-overlapping parts via ILP/MIP with cycle cuts. 605 allparts (list[frozenset[int]]): Candidate parts (gate sets). 606 g (dict[int, set[int]]): Gate DAG (u -> successors v). 607 weighted_info (tuple | None): Optional tuple providing cost modeling: 608 (single_qubit_chains, max_qubits_per_partition, go, rgo, gate_to_qubit, gate_to_tqubit). 609 gurobi_direct (bool): If True, build and solve directly in gurobipy with callbacks. 610 use_order (bool): If True, add an ordering formulation to avoid cycles. 613 tuple: (selected_parts, fusion_info or None) 614 selected_parts (list[frozenset[int]]): Chosen parts minimizing the cost. 615 fusion_info (tuple[set[int], set[int]] | None): (inpre, inpost) sets for 616 chain placements when weighted_info enables fusion costs. 618 if weighted_info
is not None:
619 single_qubit_chains, max_qubits_per_partition, go, rgo, gate_to_qubit, gate_to_tqubit = weighted_info
620 ignored_chains = {x
for x
in single_qubit_chains
if False}
621 single_qubit_chains_pre = {x[0]: x
for x
in single_qubit_chains
if rgo[x[0]]
and not x
in ignored_chains}
622 single_qubit_chains_post = {x[-1]: x
for x
in single_qubit_chains
if go[x[-1]]
and not x
in ignored_chains}
623 single_qubit_chains_prepost = {x[0]: x
for x
in single_qubit_chains
if x[0]
in single_qubit_chains_pre
and x[-1]
in single_qubit_chains_post}
624 def fortet_inequalities(x, y, z):
625 return [z-x<=0, z-y<=0, x+y-z<=1]
627 gate_to_parts = {x: []
for x
in g}
628 for i, part
in enumerate(allparts):
629 for gate
in part: gate_to_parts[gate].append(i)
631 from gurobipy
import Env, Model, GRB
632 import gurobipy
as gp
634 env.setParam(
"OutputFlag", 0)
635 with Model(env=env)
as m:
636 m.setParam(GRB.Param.IntegralityFocus, 1)
637 m.setParam(GRB.Param.LazyConstraints, 1)
638 x = m.addVars(range(N), lb=[0]*N, ub=[1]*N, vtype=[GRB.BINARY]*N, name=[
"x_" + str(i)
for i
in range(N)])
639 for i
in g: m.addConstr(gp.quicksum(x[j]
for j
in gate_to_parts[i]) == 1)
640 if weights
is not None: m.setObjective(gp.quicksum((weights[i]*N+1) * x[i]
for i
in range(N)), GRB.MINIMIZE)
641 elif weighted_info
is None: m.setObjective(gp.quicksum(x[i]
for i
in range(N)), GRB.MINIMIZE)
643 Npre, Npost, Nprepost = len(single_qubit_chains_pre), len(single_qubit_chains_post), len(single_qubit_chains_prepost)
644 pre = m.addVars(list(single_qubit_chains_pre), lb=[0]*Npre, ub=[1]*Npre, vtype=[GRB.BINARY]*Npre, name=[
"pre_" + str(i)
for i
in single_qubit_chains_pre])
645 post = m.addVars(list(single_qubit_chains_post), lb=[0]*Npost, ub=[1]*Npost, vtype=[GRB.BINARY]*Npost, name=[
"post_" + str(i)
for i
in single_qubit_chains_post])
646 noprepost = m.addVars(list(single_qubit_chains_prepost), lb=[0]*Nprepost, ub=[1]*Nprepost, vtype=[GRB.BINARY]*Nprepost, name=[
"prepost_" + str(i)
for i
in single_qubit_chains_prepost])
648 S, t, u, targets = [], {}, {}, {}
649 surrounded = {s: []
for s
in noprepost}
652 surrounded_chains = {t
for s
in part
for t
in go[s]
if t
in single_qubit_chains_prepost
and go[single_qubit_chains_prepost[t][-1]]
and next(iter(go[single_qubit_chains_prepost[t][-1]]))
in part}
653 for v
in surrounded_chains: surrounded[v].append(i)
654 fullpart = frozenset.union(part, *(single_qubit_chains_prepost[v]
for v
in surrounded_chains))
655 qubits = set.union(*(gate_to_qubit[v]
for v
in fullpart))
656 tqubits = {gate_to_tqubit[v]
for v
in fullpart}
657 cqubits = qubits - tqubits
660 is_pure = len({frozenset(gate_to_qubit[x]-{gate_to_tqubit[x]})
for x
in fullpart}) == 1
663 if s
in single_qubit_chains_pre:
664 v = gate_to_tqubit[s]
666 a = m.addVar(lb = 0, ub = 1, vtype = GRB.BINARY, name=f
"pre_t_{i}_{s}")
668 for z
in fortet_inequalities(pre[s], x[i], a): m.addConstr(z)
669 targets[i].setdefault(v, []).append(a)
670 elif not s
in fullpart:
671 if is_pure: impurities.append(pre[s])
673 if s
in noprepost: m.addConstr(pre[s] + post[single_qubit_chains_prepost[s][-1]] >= x[i])
674 else: m.addConstr(pre[s] >= x[i])
676 if s
in single_qubit_chains_post:
677 v = gate_to_tqubit[s]
679 a = m.addVar(lb = 0, ub = 1, vtype = GRB.BINARY, name=f
"post_t_{i}_{s}")
681 for z
in fortet_inequalities(post[s], x[i], a): m.addConstr(z)
682 targets[i].setdefault(v, []).append(a)
683 elif not s
in fullpart:
684 if is_pure: impurities.append(post[s])
686 if s
in noprepost: m.addConstr(post[s] + pre[single_qubit_chains_prepost[s][0]] >= x[i])
687 else: m.addConstr(post[s] >= x[i])
688 Nu = len(targets[i]); Nt = Nu+1
689 t[i] = m.addVars(range(Nt), lb=[0]*Nt, ub=[1]*Nt, vtype=[GRB.BINARY]*Nt, name=[f
"t_{i}_" + str(j)
for j
in range(Nt)])
690 u[i] = m.addVars(list(targets[i]), lb=[0]*Nu, ub=[1]*Nu, vtype=[GRB.BINARY]*Nu, name=[f
"u_{i}_" + str(j)
for j
in targets[i]])
691 for s
in u[i]: m.addConstr(u[i][s] <= x[i])
692 for target
in targets[i]:
693 for a
in targets[i][target]: m.addConstr(u[i][target] >= a)
694 m.addConstr(gp.quicksum(t[i][j]
for j
in range(Nt)) == x[i])
695 m.addConstr(gp.quicksum(j*t[i][j]
for j
in range(Nt)) == gp.quicksum(u[i][s]
for s
in u[i]))
696 if is_pure
and impurities:
697 isimpure = m.addVar(lb=0, ub=1, vtype=GRB.BINARY, name=f
"impurity_{i}")
698 for z
in impurities: m.addConstr(isimpure >= z)
699 gate_qubits = len(qubits)
701 control_qubits = len(cqubits) - j
702 g_size = 2**(gate_qubits-control_qubits)
703 if is_pure
and j==0
and impurities:
704 t0 = m.addVars(range(2), lb=[0]*2, ub=[1]*2, vtype=[GRB.BINARY]*2, name=[f
"t0_{i}_" + str(k)
for k
in range(2)])
705 for z
in fortet_inequalities(t[i][j], isimpure, t0[0]): m.addConstr(z)
706 for z
in fortet_inequalities(t[i][j], 1-isimpure, t0[1]): m.addConstr(z)
707 S.append(t0[0] * (2**max_qubits_per_partition * (g_size * (4 + 2) + 2 * (g_size - 1))))
708 S.append(t0[1] * (2**(max_qubits_per_partition-control_qubits) * (g_size * (4 + 2) + 2 * (g_size - 1))))
710 S.append(t[i][j] * (2**(max_qubits_per_partition-(control_qubits
if is_pure
and j==0
else 0)) * (g_size * (4 + 2) + 2 * (g_size - 1))))
712 if not s
in noprepost:
713 S.append((1-pre[s])*(2**max_qubits_per_partition * (2 * (4 + 2) + 2)))
715 m.addConstr(pre[s] + post[single_qubit_chains_prepost[s][-1]] + noprepost[s] + gp.quicksum(x[i]
for i
in surrounded[s]) == 1)
716 S.append(noprepost[s]*(2**max_qubits_per_partition * (2 * (4 + 2) + 2)))
718 if not single_qubit_chains_post[s][0]
in noprepost:
719 S.append((1-post[s])*(2**max_qubits_per_partition * (2 * (4 + 2) + 2)))
720 m.setObjective(gp.quicksum(S)*N+gp.quicksum(x[i]
for i
in range(N)), GRB.MINIMIZE)
722 if where == GRB.Callback.MIPSOL:
723 x_val = m.cbGetSolution([x[i]
for i
in range(N)])
724 badsccs =
sol_to_badsccs(g, allparts, [i
for i, xv
in enumerate(x_val)
if int(round(xv))])
725 for badscc
in badsccs:
727 m.cbLazy(gp.quicksum(x[j]
for j
in badscc) <= len(badscc) - 1)
729 if m.status == GRB.OPTIMAL:
730 return [i
for i
in range(N)
if int(round(x[i].getAttr(
"X")))], ({i
for i
in pre
if int(round(pre[i].getAttr(
"X")))}, {i
for i
in post
if int(round(post[i].getAttr(
"X")))})
if weighted_info
is not None and single_qubit_chains
is not None else None 732 prob = pulp.LpProblem(
"OptimalPartitioning", pulp.LpMinimize)
733 x = pulp.LpVariable.dicts(
"x", (i
for i
in range(N)), cat=
"Binary")
734 for i
in g: prob += pulp.lpSum(x[j]
for j
in gate_to_parts[i]) == 1
736 order = pulp.LpVariable.dicts(
"ord", (i
for i
in range(N)), lowBound=0, upBound=N-1, cat=
"Continuous")
740 prob += order[u] + 1 <= order[v] + N*(1-x[u]+1-x[v])
745 if weights
is not None: prob.setObjective(pulp.lpSum((weights[i]*N+1) * x[i]
for i
in range(N)))
746 elif weighted_info
is None: prob.setObjective(pulp.lpSum(x[i]
for i
in range(N)))
748 Npre, Npost, Nprepost = len(single_qubit_chains_pre), len(single_qubit_chains_post), len(single_qubit_chains_prepost)
749 pre = pulp.LpVariable.dicts(
"pre", list(single_qubit_chains_pre), cat=
"Binary")
750 post = pulp.LpVariable.dicts(
"post", list(single_qubit_chains_post), cat=
"Binary")
751 noprepost = pulp.LpVariable.dicts(
"prepost", list(single_qubit_chains_prepost), cat=
"Binary")
752 S, t, u, targets = [], {}, {}, {}
753 surrounded = {s: []
for s
in noprepost}
756 surrounded_chains = {t
for s
in part
for t
in go[s]
if t
in single_qubit_chains_prepost
and go[single_qubit_chains_prepost[t][-1]]
and next(iter(go[single_qubit_chains_prepost[t][-1]]))
in part}
757 for v
in surrounded_chains: surrounded[v].append(i)
758 fullpart = frozenset.union(part, *(single_qubit_chains_prepost[v]
for v
in surrounded_chains))
759 qubits = set.union(*(gate_to_qubit[v]
for v
in fullpart))
760 tqubits = {gate_to_tqubit[v]
for v
in fullpart}
761 cqubits = qubits - tqubits
764 is_pure = len({frozenset(gate_to_qubit[x]-{gate_to_tqubit[x]})
for x
in fullpart}) == 1
767 if s
in single_qubit_chains_pre:
768 v = gate_to_tqubit[s]
770 a = pulp.LpVariable(f
"pre_t_{i}_{s}", cat=
"Binary")
771 for z
in fortet_inequalities(pre[s], x[i], a): prob += z
772 targets[i].setdefault(v, []).append(a)
773 elif not s
in fullpart:
774 if is_pure: impurities.append(pre[s])
776 if s
in noprepost: prob += pre[s] + post[single_qubit_chains_prepost[s][-1]] >= x[i]
777 else: prob += pre[s] >= x[i]
779 if s
in single_qubit_chains_post:
780 v = gate_to_tqubit[s]
782 a = pulp.LpVariable(f
"post_t_{i}_{s}", cat=
"Binary")
783 for z
in fortet_inequalities(post[s], x[i], a): prob += z
784 targets[i].setdefault(v, []).append(a)
785 elif not s
in fullpart:
786 if is_pure: impurities.append(post[s])
788 if s
in noprepost: prob += post[s] + pre[single_qubit_chains_prepost[s][0]] >= x[i]
789 else: prob += post[s] >= x[i]
790 Nu = len(targets[i]); Nt = Nu+1
791 t[i] = pulp.LpVariable.dicts(f
"t_{i}", range(Nt), cat=
"Binary")
792 u[i] = pulp.LpVariable.dicts(f
"u_{i}", list(targets[i]), cat=
"Binary")
793 for s
in u[i]: prob += u[i][s] <= x[i]
794 for target
in targets[i]:
795 for a
in targets[i][target]: prob += u[i][target] >= a
796 prob += pulp.lpSum(t[i][j]
for j
in range(Nt)) == x[i]
797 prob += pulp.lpSum(j*t[i][j]
for j
in range(Nt)) == pulp.lpSum(u[i][s]
for s
in u[i])
798 if is_pure
and impurities:
799 isimpure = pulp.LpVariable(f
"impurity_{i}", cat=
"Binary")
800 for z
in impurities: prob += isimpure >= z
801 gate_qubits = len(qubits)
803 control_qubits = len(cqubits) - j
804 g_size = 2**(gate_qubits-control_qubits)
805 if is_pure
and j==0
and impurities:
806 t0 = pulp.LpVariable.dicts(f
"t0_{i}", range(2), cat=
"Binary")
807 for z
in fortet_inequalities(t[i][j], isimpure, t0[0]): prob += z
808 for z
in fortet_inequalities(t[i][j], 1-isimpure, t0[1]): prob += z
809 S.append(t0[0] * (2**max_qubits_per_partition * (g_size * (4 + 2) + 2 * (g_size - 1))))
810 S.append(t0[1] * (2**(max_qubits_per_partition-control_qubits) * (g_size * (4 + 2) + 2 * (g_size - 1))))
812 S.append(t[i][j] * (2**(max_qubits_per_partition-(control_qubits
if is_pure
and j==0
else 0)) * (g_size * (4 + 2) + 2 * (g_size - 1))))
814 if not s
in noprepost:
815 S.append((1-pre[s])*(2**max_qubits_per_partition * (2 * (4 + 2) + 2)))
817 prob += pre[s] + post[single_qubit_chains_prepost[s][-1]] + noprepost[s] + pulp.lpSum(x[i]
for i
in surrounded[s]) == 1
818 S.append(noprepost[s]*(2**max_qubits_per_partition * (2 * (4 + 2) + 2)))
820 if not single_qubit_chains_post[s][0]
in noprepost:
821 S.append((1-post[s])*(2**max_qubits_per_partition * (2 * (4 + 2) + 2)))
822 prob.setObjective(pulp.lpSum(S)*N+pulp.lpSum(x[i]
for i
in range(N)))
828 from gurobipy
import GRB
829 import gurobipy
as gp
831 if where == GRB.Callback.MIPSOL:
832 xg = [m.getVarByName(x[i].name)
for i
in range(N)]
833 x_val = m.cbGetSolution([xg[i]
for i
in range(N)])
834 badsccs =
sol_to_badsccs(g, allparts, [i
for i, xv
in enumerate(x_val)
if int(round(xv))])
835 for badscc
in badsccs:
837 m.cbLazy(gp.quicksum(xg[j]
for j
in badscc) <= len(badscc) - 1)
838 except Exception
as exc:
848 prob.solve(pulp.PULP_CBC_CMD(msg=
False))
851 L = [i
for i
in range(N)
if int(round(pulp.value(x[i])))]
853 if not badsccs:
break 854 for badscc
in badsccs:
856 prob += pulp.lpSum(x[j]
for j
in badscc) <= len(badscc) - 1
857 return L, ({i
for i
in pre
if int(pulp.value(pre[i]))}, {i
for i
in post
if int(pulp.value(post[i]))})
if weighted_info
is not None and single_qubit_chains
is not None else None 861 Generate all feasible parts (gate sets) for partitioning a circuit. 864 c: SQUANDER circuit object. 865 max_qubits_per_partition (int): Max allowed qubits per partition. 867 tuple: (allparts, go, rgo, single_qubit_chains, gate_to_qubit, gate_to_tqubit) 868 allparts (set[frozenset[int]]): All feasible parts (gate sets). 869 g (dict[int, set[int]]): Gate -> successors (original graph). 870 go (dict[int, set[int]]): Gate -> successors. 871 rgo (dict[int, set[int]]): Gate -> predecessors. 872 single_qubit_chains (set[tuple[int]] | None): Set of single-qubit chains (as tuples). 873 gate_to_qubit (dict[int, set[int]]): Gate -> qubits acted on. 874 gate_to_tqubit (dict[int, int | None]): Gate -> target qubit (or None). 877 gate_to_tqubit = { i: g.get_Target_Qbit()
for i, g
in gate_dict.items() }
880 topo_index = {x: i
for i, x
in enumerate(topo_order)}
883 def bfs_reach(X, g, rg, reach, Q):
884 level, nextlevel, visited = set(X), set(), set()
889 if not R <= visited:
continue 890 Qs[v] = Q
if v
in X
else set.union(gate_to_qubit[v], *(Qs[u]
for u
in R))
891 if len(Qs[v]) <= max_qubits_per_partition:
892 nextlevel |= g[v] & reach
895 level, nextlevel = nextlevel, level
901 X, Y, Q = {t}, set(topo_order[topo_index[t]+1:]), gate_to_qubit[t]
902 Anew, Bnew = reach[t] & Y, revreach[t] & Y
904 prune = Anew - bfs_reach(X, g, rg, Anew, Q)
905 Y -= prune; Anew -= prune
906 stack = [({t}, Y, Anew, Bnew, list(sorted(Anew, key=topo_index.__getitem__)), list(sorted(Bnew, key=topo_index.__getitem__, reverse=
True)), Q)]
908 X, Y, A, B, As, Bs, Q = stack.pop()
912 R = A & revreach[v]; R.add(v)
916 R = B & reach[v]; R.add(v)
919 allparts.add(frozenset(X));
continue 921 stack.append((X, Y, A, B, As, Bs, Q))
924 newQ |= gate_to_qubit[x]
925 if len(newQ) > max_qubits_per_partition:
break 927 Xnew, Ynew, Anew, Bnew = X | R, Y - R, A - R, B - R
928 for x
in R: Anew |= reach[x] & Ynew; Bnew |= revreach[x] & Ynew
929 prune = Anew - bfs_reach(Xnew, g, rg, Anew, newQ)
930 Ynew -= prune; Anew -= prune
931 prune = Bnew - bfs_reach(Xnew, rg, g, Bnew, newQ)
932 Ynew -= prune; Bnew -= prune
933 stack.append((Xnew, Ynew, Anew, Bnew, list(sorted(Anew, key=topo_index.__getitem__)), list(sorted(Bnew, key=topo_index.__getitem__, reverse=
True)), newQ))
934 return list(allparts), g, go, rgo, single_qubit_chains, gate_to_qubit, gate_to_tqubit
935 def max_partitions(c, max_qubits_per_partition, use_ilp=True, fusion_cost=False, control_aware=False):
937 Enumerate feasible parts and select a maximum/optimal partitioning of a circuit. 940 c: SQUANDER circuit object. 941 max_qubits_per_partition (int): Max allowed qubits per partition. 942 use_ilp (bool): If True, solve selection via ILP; else greedy largest-first. 943 fusion_cost (bool): If True, include FLOP-based cost with fusion/controls. 944 control_aware (bool): If True and fusion_cost, treat single-qubit chains as 945 pre/post relative to targets. 948 tuple: (partitioned_circ, param_order, parts) 949 partitioned_circ: 2-level partitioned circuit (SQUANDER object). 950 param_order (list[tuple[int,int,int]]): (source_idx, dest_idx, param_count). 951 parts (list[frozenset[int]]): Final partition assignment sets. 953 allparts, g, go, rgo, single_qubit_chains, gate_to_qubit, gate_to_tqubit =
get_all_partitions(c, max_qubits_per_partition)
955 weights =
parts_to_float_ops(max_qubits_per_partition, gate_to_qubit,
None, allparts)
if fusion_cost
and not control_aware
else None 956 L, fusion_info =
ilp_global_optimal(allparts, g, (single_qubit_chains, max_qubits_per_partition, go, rgo, gate_to_qubit, gate_to_tqubit)
if control_aware
else None, weights=weights)
958 L, excluded = [], set()
960 for i
in sorted(range(len(allparts)), key=
lambda i: (len(allparts[i]), -weights[i]), reverse=
True):
961 if len(allparts[i] & excluded) != 0:
continue 963 excluded |= allparts[i]
970 from squander.partitioning.kahn
import kahn_partition_preparts
976 Quick test harness for partitioning on sample QASM circuits. 989 filename =
"benchmarks/partitioning/test_circuit/con1_216_squander.qasm" 993 from squander
import utils
995 circ, parameters = utils.qasm_to_squander_circuit(filename)
996 gate_dict = {i: gate
for i, gate
in enumerate(circ.get_Gates())}
997 gate_to_qubit = { i:
get_qubits(g)
for i, g
in gate_dict.items() }
998 gate_to_tqubit = { i: g.get_Target_Qbit()
for i, g
in gate_dict.items() }
1000 for partition
in (
max_partitions(circ, K,
False,
False),
max_partitions(circ, K,
False,
True),
max_partitions(circ, K,
True,
False),
max_partitions(circ, K,
True,
True),
max_partitions(circ, K,
True,
True,
True)):
1001 print(partition[2], len(partition[2]),
total_float_ops(circ.get_Qbit_Num(), K, gate_to_qubit,
None, partition[2]),
1002 total_float_ops(circ.get_Qbit_Num(), K, gate_to_qubit, gate_to_tqubit, partition[2]))
1004 if __name__ ==
"__main__":
def _check_gurobi_available()
def _solve_pulp_with_gurobi_or_cbc(prob, pulp, callback=None, gurobi_kwargs)
def contract_single_qubit_chains(go, rgo, gate_to_qubit, topo_order)
def max_partitions(c, max_qubits_per_partition, use_ilp=True, fusion_cost=False, control_aware=False)
def sol_to_badsccs(g, allparts, L)
def topo_sort_partitions(c, parts)
def scc_tarjan_iterative(succ)
def kahn_partition_preparts(c, max_qubit, preparts)
def two_cycles_from_dag_edges(g, gate_to_parts, allparts)
def find_next_biggest_partition(c, max_qubits_per_partition, prevparts=None)
def get_part_cycle_graph(g, gate_to_parts)
def all_cycles_from_dag_edges(succ, max_len=5)
def nuutila_reach_scc(succ, subg=None)
def get_all_partitions(c, max_qubits_per_partition)
def _get_topo_order(g, rg, gate_to_qubit)
def ilp_global_optimal(allparts, g, weighted_info=None, gurobi_direct=False, use_order=False, weights=None)
def ilp_max_partitions(c, max_qubits_per_partition)
def _print_gurobi_fallback_warning(exc)
def recombine_single_qubit_chains(g, rg, single_qubit_chains, gate_to_tqubit, L, fusion_info, surrounded_only=False)