Skip to content

lineprog

LPProblem dataclass 

LPProblem(
    constraints_eqz: list[Expression] = list(),
    constraints_gez: list[Expression] = list(),
    linear_objective: Expression = (
        lambda: Expression({})
    )(),
    quadratic_objective: list[Expression] = list(),
)

add_abs 

add_abs(expr: Expression)

Use a slack variable to add an absolute value constraint.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/lineprog.py 
166
167
168
169
170
171
172
173
174
def add_abs(self, expr: Expression):
    """
    Use a slack variable to add an absolute value constraint.
    """
    slack = Variable()
    self.add_gez(slack - expr)
    self.add_gez(slack + expr)
    self.add_gez(1.0 * slack)
    self.add_linear(1.0 * slack)

noise

conflict_graph

OneZoneConflictGraph 

OneZoneConflictGraph(moment: Moment)

Representation of the AOD conflict graph for qubits to more to their entangling partners in a single zone setup.

Assumes the qubits are specified as cirq.GridQubits with a chosen geometry.

:param moment: A cirq.Moment object containing operations (gates) to be analyzed.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/noise/conflict_graph.py 
12
13
14
15
16
17
18
19
20
def __init__(self, moment: cirq.Moment):
    """
    Initializes the conflict graph for a given moment of a cirq circuit.

    :param moment: A cirq.Moment object containing operations (gates) to be analyzed.
    """

    self.moment = moment
    self.gates_in_moment = [op for op in moment.operations if len(op.qubits) == 2]

get_move_schedule 

get_move_schedule(mover_limit: int = 10000)

Generates a move schedule by coloring the conflict graph greedily, first coloring nodes of highest degree.

Qubits that are the arguments of a single CZ gate are 'partners'. Only one partner need be moved to arrange the atoms for the 2Q gate. Thus, in coloring the conflict graph, as soon as one partner is colored, the other can be disregarded for the purpose of coloring the rest of the graph.

This sets the self.move_schedule attribute, which is a dictionary where the keys are the indices of the move moments.

:param mover_limit: The maximum number of qubits that can be moved in a single moment. Added as a constraint when coloring the conflict graph. :returns a dictionary of idx:[cirq.Qid] where idx indexes the move moment where the list of qubits move.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/noise/conflict_graph.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def get_move_schedule(self, mover_limit: int = 10000):
    """Generates a move schedule by coloring the conflict graph greedily, first coloring nodes of highest degree.

    Qubits that are the arguments of a single CZ gate are 'partners'. Only one partner need be moved to arrange the
    atoms for the 2Q gate. Thus, in coloring the conflict graph, as soon as one partner is colored, the other can be
    disregarded for the purpose of coloring the rest of the graph.

    This sets the self.move_schedule attribute, which is a dictionary where the keys are the indices of the move moments.

    :param mover_limit: The maximum number of qubits that can be moved in a single moment. Added as a constraint
        when coloring the conflict graph.
    :returns a dictionary of idx:[cirq.Qid] where idx indexes the move moment where the list of qubits move.
    """

    self._get_nodes()
    self._get_edges()
    self._get_node_degrees()

    self.ordered_nodes = sorted(self.degrees, key=self.degrees.get, reverse=True)

    move_schedule = {}
    colored_nodes = set()
    partner_node = None
    for node in self.ordered_nodes:
        colored = False
        for gate in self.gates_in_moment:
            if node in gate.qubits:
                partners = set(gate.qubits)
                partners.remove(node)
                partner_node = list(partners)[0]
        if node in colored_nodes:
            # NOTE: if a node is colored, both it and its partner are added to colored_nodes
            continue
        else:
            connected_nodes = set()
            for edge in self.edges:
                if node in edge:
                    connected_nodes.add(edge[0])
                    connected_nodes.add(edge[1])
            connected_nodes.remove(node)
            for color in move_schedule.keys():
                has_colored_neighbor = False
                for connected_node in connected_nodes:
                    # NOTE: loop through to make sure none of the connected nodes are already assigned to color.
                    if connected_node in move_schedule[color]:
                        has_colored_neighbor = True
                        break
                    else:
                        continue
                mover_limit_reached = len(move_schedule[color]) >= mover_limit
                if not (has_colored_neighbor or mover_limit_reached):
                    # NOTE: node needs color
                    move_schedule[color].add(node)
                    colored = True

                    # NOTE: add this node and it's partner to the solved nodes.
                    colored_nodes.add(node)

                    if partner_node is not None:
                        colored_nodes.add(partner_node)
                    break
            if not colored:
                move_schedule[len(move_schedule)] = {node}
                colored = True
                colored_nodes.add(node)

                if partner_node is not None:
                    colored_nodes.add(partner_node)

    assert set(colored_nodes) == set(self.ordered_nodes)

    self.move_schedule = move_schedule

    return self.move_schedule

model

GeminiNoiseModelABC dataclass 

GeminiNoiseModelABC(
    check_input_circuit: bool = True,
    *,
    local_px: float = 0.0004102,
    local_py: float = 0.0004102,
    local_pz: float = 0.0004112,
    local_loss_prob: float = 0.0,
    local_unaddressed_px: float = 2e-07,
    local_unaddressed_py: float = 2e-07,
    local_unaddressed_pz: float = 1.2e-06,
    local_unaddressed_loss_prob: float = 0.0,
    global_px: float = 6.5e-05,
    global_py: float = 6.5e-05,
    global_pz: float = 6.5e-05,
    global_loss_prob: float = 0.0,
    cz_paired_gate_px: float = 0.0006549,
    cz_paired_gate_py: float = 0.0006549,
    cz_paired_gate_pz: float = 0.003184,
    cz_gate_loss_prob: float = 0.0,
    cz_unpaired_gate_px: float = 0.0005149,
    cz_unpaired_gate_py: float = 0.0005149,
    cz_unpaired_gate_pz: float = 0.002185,
    cz_unpaired_loss_prob: float = 0.0,
    mover_px: float = 0.000806,
    mover_py: float = 0.000806,
    mover_pz: float = 0.002458,
    move_loss_prob: float = 0.0,
    sitter_px: float = 0.0003066,
    sitter_py: float = 0.0003066,
    sitter_pz: float = 0.0004639,
    sit_loss_prob: float = 0.0
)

Bases: NoiseModel, MoveNoiseModelABC

Abstract base class for all Gemini noise models.

check_input_circuit class-attribute instance-attribute 

check_input_circuit: bool = True

Determine whether or not to verify that the circuit only contains native gates.

Caution: Disabling this for circuits containing non-native gates may lead to incorrect results!

parallel_cz_errors 

parallel_cz_errors(
    ctrls: list[int], qargs: list[int], rest: list[int]
) -> dict[tuple[float, float, float, float], list[int]]

Takes a set of ctrls and qargs and returns a noise model for all qubits.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/noise/model.py 
45
46
47
48
49
50
def parallel_cz_errors(
    self, ctrls: list[int], qargs: list[int], rest: list[int]
) -> dict[tuple[float, float, float, float], list[int]]:
    raise NotImplementedError(
        "This noise model doesn't support rewrites on bloqade kernels, but should be used with cirq."
    )

GeminiOneZoneNoiseModel dataclass 

GeminiOneZoneNoiseModel(
    check_input_circuit: bool = True,
    parallelize_circuit: bool = False,
    *,
    local_px: float = 0.0004102,
    local_py: float = 0.0004102,
    local_pz: float = 0.0004112,
    local_loss_prob: float = 0.0,
    local_unaddressed_px: float = 2e-07,
    local_unaddressed_py: float = 2e-07,
    local_unaddressed_pz: float = 1.2e-06,
    local_unaddressed_loss_prob: float = 0.0,
    global_px: float = 6.5e-05,
    global_py: float = 6.5e-05,
    global_pz: float = 6.5e-05,
    global_loss_prob: float = 0.0,
    cz_paired_gate_px: float = 0.0006549,
    cz_paired_gate_py: float = 0.0006549,
    cz_paired_gate_pz: float = 0.003184,
    cz_gate_loss_prob: float = 0.0,
    cz_unpaired_gate_px: float = 0.0005149,
    cz_unpaired_gate_py: float = 0.0005149,
    cz_unpaired_gate_pz: float = 0.002185,
    cz_unpaired_loss_prob: float = 0.0,
    mover_px: float = 0.000806,
    mover_py: float = 0.000806,
    mover_pz: float = 0.002458,
    move_loss_prob: float = 0.0,
    sitter_px: float = 0.0003066,
    sitter_py: float = 0.0003066,
    sitter_pz: float = 0.0004639,
    sit_loss_prob: float = 0.0
)

Bases: GeminiOneZoneNoiseModelABC

A Cirq-compatible noise model for a one-zone implementation of the Gemini architecture.

This model introduces custom asymmetric depolarizing noise for both single- and two-qubit gates depending on whether operations are global, local, or part of a CZ interaction. Since the model assumes all atoms are in the entangling zone, error are applied that stem from application of Rydberg error, even for qubits not actively involved in a gate operation.

noisy_moment 

noisy_moment(moment: Moment, system_qubits: Sequence[Qid])

Applies a structured noise model to a given moment depending on the type of operations it contains.

For single-qubit moments
  • If all gates are identical and act on all qubits, global noise is applied.
  • Otherwise, local depolarizing noise is applied per qubit.
For two-qubit moments
  • Applies move error to move control qubits to target qubits before the gate and again to move back after the gate.
  • Applies gate error to control and target qubits.
  • Adds 1q asymmetric noise to qubits that do not participate in a gate.

Parameters:

Name Type Description Default
moment Moment

A cirq.Moment containing the original quantum operations.

 required
system_qubits Sequence[Qid]

All qubits in the system (used to determine idleness and global operations).

 required

Returns:

Type Description

A list of cirq.Moment objects: [pre-gate move noise, original moment, post-gate move noise, gate noise moment]

Raises:

Type Description
ValueError

If the moment contains multi-qubit gates involving >2 qubits, which are unsupported.
Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/noise/model.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def noisy_moment(self, moment: cirq.Moment, system_qubits: Sequence[cirq.Qid]):
    """
    Applies a structured noise model to a given moment depending on the type of operations it contains.

    For single-qubit moments:
        - If all gates are identical and act on all qubits, global noise is applied.
        - Otherwise, local depolarizing noise is applied per qubit.

    For two-qubit moments:
        - Applies move error to move control qubits to target qubits before the gate and again to move back after
            the gate.
        - Applies gate error to control and target qubits.
        - Adds 1q asymmetric noise to qubits that do not participate in a gate.

    Args:
        moment: A cirq.Moment containing the original quantum operations.
        system_qubits: All qubits in the system (used to determine idleness and global operations).

    Returns:
        A list of cirq.Moment objects:
            [pre-gate move noise, original moment, post-gate move noise, gate noise moment]

    Raises:
        ValueError: If the moment contains multi-qubit gates involving >2 qubits, which are unsupported.
    """
    # Moment with original ops
    original_moment = moment

    # Check if the moment is empty
    if len(moment.operations) == 0:
        move_noise_ops = []
        gate_noise_ops = []
    # Check if the moment contains 1-qubit gates or 2-qubit gates
    elif len(moment.operations[0].qubits) == 1:
        gate_noise_ops, move_noise_ops = self._single_qubit_moment_noise_ops(
            moment, system_qubits
        )
    elif len(moment.operations[0].qubits) == 2:
        # Check if the moment only contains two qubit gates
        assert np.all([len(op.qubits) == 2 for op in moment.operations])

        control_qubits = [op.qubits[0] for op in moment.operations]
        target_qubits = [op.qubits[1] for op in moment.operations]
        gated_qubits = control_qubits + target_qubits
        idle_atoms = list(set(system_qubits) - set(gated_qubits))

        move_noise_ops = [
            cirq.asymmetric_depolarize(*self.mover_pauli_rates).on_each(
                control_qubits
            ),
            cirq.asymmetric_depolarize(*self.sitter_pauli_rates).on_each(
                target_qubits + idle_atoms
            ),
        ]  # In this setting, we assume a 1 zone scheme where the controls move to the targets.

        gate_noise_ops = [
            cirq.asymmetric_depolarize(*self.cz_paired_pauli_rates).on_each(
                gated_qubits
            ),
            cirq.asymmetric_depolarize(*self.cz_unpaired_pauli_rates).on_each(
                idle_atoms
            ),
        ]  # In this 1 zone scheme, all unpaired atoms are in the entangling zone.
    else:
        raise ValueError(
            "Moment contains operations with more than 2 qubits, which is not supported. "
            "Correlated measurements should be added after the noise model is applied."
        )

    if move_noise_ops == []:
        move_noise_moments = []
    else:
        move_noise_moments = [cirq.Moment(move_noise_ops)]
    gate_noise_moment = cirq.Moment(gate_noise_ops)

    return [
        *move_noise_moments,
        original_moment,
        gate_noise_moment,
        *move_noise_moments,
    ]

GeminiOneZoneNoiseModelABC dataclass 

GeminiOneZoneNoiseModelABC(
    check_input_circuit: bool = True,
    parallelize_circuit: bool = False,
    *,
    local_px: float = 0.0004102,
    local_py: float = 0.0004102,
    local_pz: float = 0.0004112,
    local_loss_prob: float = 0.0,
    local_unaddressed_px: float = 2e-07,
    local_unaddressed_py: float = 2e-07,
    local_unaddressed_pz: float = 1.2e-06,
    local_unaddressed_loss_prob: float = 0.0,
    global_px: float = 6.5e-05,
    global_py: float = 6.5e-05,
    global_pz: float = 6.5e-05,
    global_loss_prob: float = 0.0,
    cz_paired_gate_px: float = 0.0006549,
    cz_paired_gate_py: float = 0.0006549,
    cz_paired_gate_pz: float = 0.003184,
    cz_gate_loss_prob: float = 0.0,
    cz_unpaired_gate_px: float = 0.0005149,
    cz_unpaired_gate_py: float = 0.0005149,
    cz_unpaired_gate_pz: float = 0.002185,
    cz_unpaired_loss_prob: float = 0.0,
    mover_px: float = 0.000806,
    mover_py: float = 0.000806,
    mover_pz: float = 0.002458,
    move_loss_prob: float = 0.0,
    sitter_px: float = 0.0003066,
    sitter_py: float = 0.0003066,
    sitter_pz: float = 0.0004639,
    sit_loss_prob: float = 0.0
)

Bases: GeminiNoiseModelABC

Abstract base class for all one-zone Gemini noise models.

noisy_moments 

noisy_moments(
    moments: Iterable[Moment], system_qubits: Sequence[Qid]
) -> Sequence[cirq.OP_TREE]

Adds possibly stateful noise to a series of moments.

Parameters:

Name Type Description Default
moments Iterable[Moment]

The moments to add noise to.

 required
system_qubits Sequence[Qid]

A list of all qubits in the system.

 required

Returns:

Type Description
Sequence[OP_TREE]

A sequence of OP_TREEEs, with the k'th tree corresponding to the
Sequence[OP_TREE]

noisy operations for the k'th moment.
Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/noise/model.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def noisy_moments(
    self, moments: Iterable[cirq.Moment], system_qubits: Sequence[cirq.Qid]
) -> Sequence[cirq.OP_TREE]:
    """Adds possibly stateful noise to a series of moments.

    Args:
        moments: The moments to add noise to.
        system_qubits: A list of all qubits in the system.

    Returns:
        A sequence of OP_TREEEs, with the k'th tree corresponding to the
        noisy operations for the k'th moment.
    """

    if self.check_input_circuit:
        self.validate_moments(moments)

    # Split into moments with only 1Q and 2Q gates
    moments_1q = [
        cirq.Moment([op for op in moment.operations if len(op.qubits) == 1])
        for moment in moments
    ]
    moments_2q = [
        cirq.Moment([op for op in moment.operations if len(op.qubits) == 2])
        for moment in moments
    ]

    assert len(moments_1q) == len(moments_2q)

    interleaved_moments = []
    for idx, moment in enumerate(moments_1q):
        interleaved_moments.append(moment)
        interleaved_moments.append(moments_2q[idx])

    interleaved_circuit = cirq.Circuit.from_moments(*interleaved_moments)

    # Combine subsequent 1Q gates
    compressed_circuit = cirq.merge_single_qubit_moments_to_phxz(
        interleaved_circuit
    )
    if self.parallelize_circuit:
        compressed_circuit = parallelize(compressed_circuit)

    return self._noisy_moments_impl_moment(
        compressed_circuit.moments, system_qubits
    )

GeminiOneZoneNoiseModelConflictGraphMoves dataclass 

GeminiOneZoneNoiseModelConflictGraphMoves(
    check_input_circuit: bool = True,
    parallelize_circuit: bool = False,
    max_parallel_movers: int = 10000,
    *,
    local_px: float = 0.0004102,
    local_py: float = 0.0004102,
    local_pz: float = 0.0004112,
    local_loss_prob: float = 0.0,
    local_unaddressed_px: float = 2e-07,
    local_unaddressed_py: float = 2e-07,
    local_unaddressed_pz: float = 1.2e-06,
    local_unaddressed_loss_prob: float = 0.0,
    global_px: float = 6.5e-05,
    global_py: float = 6.5e-05,
    global_pz: float = 6.5e-05,
    global_loss_prob: float = 0.0,
    cz_paired_gate_px: float = 0.0006549,
    cz_paired_gate_py: float = 0.0006549,
    cz_paired_gate_pz: float = 0.003184,
    cz_gate_loss_prob: float = 0.0,
    cz_unpaired_gate_px: float = 0.0005149,
    cz_unpaired_gate_py: float = 0.0005149,
    cz_unpaired_gate_pz: float = 0.002185,
    cz_unpaired_loss_prob: float = 0.0,
    mover_px: float = 0.000806,
    mover_py: float = 0.000806,
    mover_pz: float = 0.002458,
    move_loss_prob: float = 0.0,
    sitter_px: float = 0.0003066,
    sitter_py: float = 0.0003066,
    sitter_pz: float = 0.0004639,
    sit_loss_prob: float = 0.0
)

Bases: GeminiOneZoneNoiseModel

A Cirq noise model that uses a conflict graph to schedule moves in a one-zone Gemini architecture.

Assumes that the qubits are cirq.GridQubits, such that the assignment of row, column coordinates define the initial geometry. An SLM site at the two qubit interaction distance is also assumed next to each cirq.GridQubit to allow for multiple moves before a single Rydberg pulse is applied for a parallel CZ.

noisy_moment 

noisy_moment(moment, system_qubits)

Applies a structured noise model to a given moment depending on the type of operations it contains.

For single-qubit moments
  • If all gates are identical and act on all qubits, global noise is applied.
  • Otherwise, local depolarizing noise is applied per qubit.
For two-qubit moments
  • Applies move error to move control qubits to target qubits before the gate and again to move back after the gate.
  • Applies gate error to control and target qubits.
  • Adds 1q asymmetric noise to qubits that do not participate in a gate.

Parameters:

Name Type Description Default
moment Moment

A cirq.Moment containing the original quantum operations.

 required
system_qubits Sequence[Qid]

All qubits in the system (used to determine idleness and global operations).

 required

Returns:

Type Description

A list of cirq.Moment objects: [pre-gate move noise, original moment, post-gate move noise, gate noise moment]

Raises:

Type Description
ValueError

If the moment contains multi-qubit gates involving >2 qubits, which are unsupported.
Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/noise/model.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
def noisy_moment(self, moment, system_qubits):
    # Moment with original ops
    original_moment = moment
    assert np.all(
        [isinstance(q, cirq.GridQubit) for q in system_qubits]
    ), "Found a qubit that is not a GridQubit."
    # Check if the moment is empty
    if len(moment.operations) == 0:
        move_moments = []
        gate_noise_ops = []
    # Check if the moment contains 1-qubit gates or 2-qubit gates
    elif len(moment.operations[0].qubits) == 1:
        gate_noise_ops, _ = self._single_qubit_moment_noise_ops(
            moment, system_qubits
        )
        move_moments = []
    elif len(moment.operations[0].qubits) == 2:
        cg = OneZoneConflictGraph(moment)
        schedule = cg.get_move_schedule(mover_limit=self.max_parallel_movers)
        move_moments = []
        for move_moment_idx, movers in schedule.items():
            control_qubits = list(movers)
            target_qubits = list(
                set(
                    [op.qubits[0] for op in moment.operations]
                    + [op.qubits[1] for op in moment.operations]
                )
                - movers
            )
            gated_qubits = control_qubits + target_qubits
            idle_atoms = list(set(system_qubits) - set(gated_qubits))

            move_noise_ops = [
                cirq.asymmetric_depolarize(*self.mover_pauli_rates).on_each(
                    control_qubits
                ),
                cirq.asymmetric_depolarize(*self.sitter_pauli_rates).on_each(
                    target_qubits + idle_atoms
                ),
            ]

            move_moments.append(cirq.Moment(move_noise_ops))

        control_qubits = [op.qubits[0] for op in moment.operations]
        target_qubits = [op.qubits[1] for op in moment.operations]
        gated_qubits = control_qubits + target_qubits
        idle_atoms = list(set(system_qubits) - set(gated_qubits))

        gate_noise_ops = [
            cirq.asymmetric_depolarize(*self.cz_paired_pauli_rates).on_each(
                gated_qubits
            ),
            cirq.asymmetric_depolarize(*self.cz_unpaired_pauli_rates).on_each(
                idle_atoms
            ),
        ]  # In this 1 zone scheme, all unpaired atoms are in the entangling zone.
    else:
        raise ValueError(
            "Moment contains operations with more than 2 qubits, which is not supported."
            "Correlated measurements should be added after the noise model is applied."
        )

    gate_noise_moment = cirq.Moment(gate_noise_ops)

    return [
        *move_moments,
        original_moment,
        gate_noise_moment,
        *(move_moments[::-1]),
    ]

GeminiOneZoneNoiseModelCorrelated dataclass 

GeminiOneZoneNoiseModelCorrelated(
    check_input_circuit: bool = True,
    parallelize_circuit: bool = False,
    cz_paired_correlated_rates: ndarray = _default_cz_paired_correlated_rates(),
    *,
    local_px: float = 0.0004102,
    local_py: float = 0.0004102,
    local_pz: float = 0.0004112,
    local_loss_prob: float = 0.0,
    local_unaddressed_px: float = 2e-07,
    local_unaddressed_py: float = 2e-07,
    local_unaddressed_pz: float = 1.2e-06,
    local_unaddressed_loss_prob: float = 0.0,
    global_px: float = 6.5e-05,
    global_py: float = 6.5e-05,
    global_pz: float = 6.5e-05,
    global_loss_prob: float = 0.0,
    cz_paired_gate_px: float = 0.0006549,
    cz_paired_gate_py: float = 0.0006549,
    cz_paired_gate_pz: float = 0.003184,
    cz_gate_loss_prob: float = 0.0,
    cz_unpaired_gate_px: float = 0.0005149,
    cz_unpaired_gate_py: float = 0.0005149,
    cz_unpaired_gate_pz: float = 0.002185,
    cz_unpaired_loss_prob: float = 0.0,
    mover_px: float = 0.000806,
    mover_py: float = 0.000806,
    mover_pz: float = 0.002458,
    move_loss_prob: float = 0.0,
    sitter_px: float = 0.0003066,
    sitter_py: float = 0.0003066,
    sitter_pz: float = 0.0004639,
    sit_loss_prob: float = 0.0
)

Bases: GeminiOneZoneNoiseModel

A Cirq noise model for implementing correlated two-qubit Pauli errors in a one-zone Gemini architecture.

noisy_moment 

noisy_moment(moment, system_qubits)

Applies a structured noise model to a given moment depending on the type of operations it contains.

For single-qubit moments
  • If all gates are identical and act on all qubits, global noise is applied.
  • Otherwise, local depolarizing noise is applied per qubit.
For two-qubit moments
  • Applies move error to move control qubits to target qubits before the gate and again to move back after the gate.
  • Applies gate error to control and target qubits.
  • Adds 1q asymmetric noise to qubits that do not participate in a gate.

Parameters:

Name Type Description Default
moment Moment

A cirq.Moment containing the original quantum operations.

 required
system_qubits Sequence[Qid]

All qubits in the system (used to determine idleness and global operations).

 required

Returns:

Type Description

A list of cirq.Moment objects: [pre-gate move noise, original moment, post-gate move noise, gate noise moment]

Raises:

Type Description
ValueError

If the moment contains multi-qubit gates involving >2 qubits, which are unsupported.
Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/noise/model.py 
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
def noisy_moment(self, moment, system_qubits):
    # Moment with original ops
    original_moment = moment

    # Check if the moment is empty
    if len(moment.operations) == 0:
        move_noise_ops = []
        gate_noise_ops = []
    # Check if the moment contains 1-qubit gates or 2-qubit gates
    elif len(moment.operations[0].qubits) == 1:
        gate_noise_ops, move_noise_ops = self._single_qubit_moment_noise_ops(
            moment, system_qubits
        )
    elif len(moment.operations[0].qubits) == 2:
        control_qubits = [op.qubits[0] for op in moment.operations]
        target_qubits = [op.qubits[1] for op in moment.operations]
        gated_qubits = control_qubits + target_qubits
        idle_atoms = list(set(system_qubits) - set(gated_qubits))

        move_noise_ops = [
            cirq.asymmetric_depolarize(*self.mover_pauli_rates).on_each(
                control_qubits
            ),
            cirq.asymmetric_depolarize(*self.sitter_pauli_rates).on_each(
                target_qubits + idle_atoms
            ),
        ]  # In this setting, we assume a 1 zone scheme where the controls move to the targets.

        # Add correlated noise channels for entangling pairs
        two_qubit_pauli = self.two_qubit_pauli
        gate_noise_ops = [
            two_qubit_pauli.on_each([c, t])
            for c, t in zip(control_qubits, target_qubits)
        ]

        # In this 1 zone scheme, all unpaired atoms are in the entangling zone.
        idle_depolarize = cirq.asymmetric_depolarize(
            *self.cz_unpaired_pauli_rates
        ).on_each(idle_atoms)

        gate_noise_ops.append(idle_depolarize)
    else:
        raise ValueError(
            "Moment contains operations with more than 2 qubits, which is not supported."
            "Correlated measurements should be added after the noise model is applied."
        )
    if move_noise_ops == []:
        move_noise_moments = []
    else:
        move_noise_moments = [cirq.Moment(move_noise_ops)]
    gate_noise_moment = cirq.Moment(gate_noise_ops)

    return [
        *move_noise_moments,
        original_moment,
        gate_noise_moment,
        *move_noise_moments,
    ]

GeminiTwoZoneNoiseModel dataclass 

GeminiTwoZoneNoiseModel(
    check_input_circuit: bool = True,
    *,
    local_px: float = 0.0004102,
    local_py: float = 0.0004102,
    local_pz: float = 0.0004112,
    local_loss_prob: float = 0.0,
    local_unaddressed_px: float = 2e-07,
    local_unaddressed_py: float = 2e-07,
    local_unaddressed_pz: float = 1.2e-06,
    local_unaddressed_loss_prob: float = 0.0,
    global_px: float = 6.5e-05,
    global_py: float = 6.5e-05,
    global_pz: float = 6.5e-05,
    global_loss_prob: float = 0.0,
    cz_paired_gate_px: float = 0.0006549,
    cz_paired_gate_py: float = 0.0006549,
    cz_paired_gate_pz: float = 0.003184,
    cz_gate_loss_prob: float = 0.0,
    cz_unpaired_gate_px: float = 0.0005149,
    cz_unpaired_gate_py: float = 0.0005149,
    cz_unpaired_gate_pz: float = 0.002185,
    cz_unpaired_loss_prob: float = 0.0,
    mover_px: float = 0.000806,
    mover_py: float = 0.000806,
    mover_pz: float = 0.002458,
    move_loss_prob: float = 0.0,
    sitter_px: float = 0.0003066,
    sitter_py: float = 0.0003066,
    sitter_pz: float = 0.0004639,
    sit_loss_prob: float = 0.0
)

Bases: GeminiNoiseModelABC

noisy_moments 

noisy_moments(
    moments: Iterable[Moment], system_qubits: Sequence[Qid]
) -> Sequence[cirq.OP_TREE]

Adds possibly stateful noise to a series of moments.

Parameters:

Name Type Description Default
moments Iterable[Moment]

The moments to add noise to.

 required
system_qubits Sequence[Qid]

A list of all qubits in the system.

 required

Returns:

Type Description
Sequence[OP_TREE]

A sequence of OP_TREEEs, with the k'th tree corresponding to the
Sequence[OP_TREE]

noisy operations for the k'th moment.
Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/noise/model.py 
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
def noisy_moments(
    self, moments: Iterable[cirq.Moment], system_qubits: Sequence[cirq.Qid]
) -> Sequence[cirq.OP_TREE]:
    """Adds possibly stateful noise to a series of moments.

    Args:
        moments: The moments to add noise to.
        system_qubits: A list of all qubits in the system.

    Returns:
        A sequence of OP_TREEEs, with the k'th tree corresponding to the
        noisy operations for the k'th moment.
    """

    if self.check_input_circuit:
        self.validate_moments(moments)

    moments = list(moments)

    if len(moments) == 0:
        return []

    nqubs = len(system_qubits)
    noisy_moment_list = []

    prev_moment: cirq.Moment | None = None

    # TODO: clean up error getters so they return a list moments rather than circuits
    for i in range(len(moments)):
        noisy_moment_list.extend(
            [
                moment
                for moment in _two_zone_utils.get_move_error_channel_two_zoned(
                    moments[i],
                    prev_moment,
                    np.array(self.mover_pauli_rates),
                    np.array(self.sitter_pauli_rates),
                    nqubs,
                ).moments
                if len(moment) > 0
            ]
        )

        noisy_moment_list.append(moments[i])

        noisy_moment_list.extend(
            [
                moment
                for moment in _two_zone_utils.get_gate_error_channel(
                    moments[i],
                    np.array(self.local_pauli_rates),
                    np.array(self.global_pauli_rates),
                    np.array(
                        self.cz_paired_pauli_rates + self.cz_paired_pauli_rates
                    ),
                    np.array(self.cz_unpaired_pauli_rates),
                ).moments
                if len(moment) > 0
            ]
        )

        prev_moment = moments[i]

    return noisy_moment_list

transform

transform_circuit 

transform_circuit(
    circuit: Circuit,
    to_native_gateset: bool = True,
    model: NoiseModel | None = None,
    parallelize_circuit: bool = False,
) -> cirq.Circuit

Transform an input circuit into one with the native gateset with noise operations added.

Noise operations will be added to all qubits in circuit.all_qubits(), regardless of whether the output of the circuit optimizers contain all the qubits.

Parameters:

Name Type Description Default
circuit Circuit

The input circuit.

 required

Other Parameters:

Name Type Description
to_native_gateset bool

Whether or not to convert the input circuit to one using the native set of gates (cirq.CZTargetGateset) only. Defaults to True. Note, that if you use an input circuit that has gates different from this gateset and don't convert it, may lead to incorrect results and errors.
model NoiseModel

The cirq noise model to apply to the circuit. Usually, you want to use one of the ones supplied in this submodule, such as GeminiOneZoneNoiseModel.
parallelize_circuit bool

Whether or not to parallelize the circuit as much as possible after it's been converted to the native gateset. Defaults to False.

Returns:

Type Description
Circuit

cirq.Circuit: The resulting noisy circuit.
Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/noise/transform.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def transform_circuit(
    circuit: cirq.Circuit,
    to_native_gateset: bool = True,
    model: cirq.NoiseModel | None = None,
    parallelize_circuit: bool = False,
) -> cirq.Circuit:
    """Transform an input circuit into one with the native gateset with noise operations added.

    Noise operations will be added to all qubits in circuit.all_qubits(), regardless of whether the output of the
    circuit optimizers contain all the qubits.

    Args:
        circuit (cirq.Circuit): The input circuit.

    Keyword Arguments:
        to_native_gateset (bool): Whether or not to convert the input circuit to one using the native set of gates (`cirq.CZTargetGateset`)
            only. Defaults to `True`. Note, that if you use an input circuit that has gates different from this gateset and don't convert it,
            may lead to incorrect results and errors.
        model (cirq.NoiseModel): The cirq noise model to apply to the circuit. Usually, you want to use one of the ones supplied in this submodule,
            such as `GeminiOneZoneNoiseModel`.
        parallelize_circuit (bool): Whether or not to parallelize the circuit as much as possible after it's been converted to the native gateset.
            Defaults to `False`.

    Returns:
        cirq.Circuit:
            The resulting noisy circuit.
    """
    if model is None:
        model = GeminiOneZoneNoiseModel(parallelize_circuit=parallelize_circuit)

    # only parallelize here if we aren't parallelizing inside a one-zone model
    parallelize_circuit_here = parallelize_circuit and not isinstance(
        model, GeminiOneZoneNoiseModelABC
    )

    system_qubits = sorted(circuit.all_qubits())
    # Transform to CZ + PhasedXZ gateset.
    if to_native_gateset and not parallelize_circuit_here:
        native_circuit = transpile(circuit)
    elif parallelize_circuit_here:
        native_circuit = parallelize(circuit)
    else:
        native_circuit = circuit

    # Add noise
    noisy_circuit = cirq.Circuit()
    for op_tree in model.noisy_moments(native_circuit, system_qubits):
        # Keep moments aligned
        noisy_circuit += cirq.Circuit(op_tree)

    return noisy_circuit

parallelize

auto_similarity 

auto_similarity(
    circuit: Circuit, weight_1q: float, weight_2q: float
) -> tuple[cirq.Circuit, dict[Hashable, float]]

Automatically tag the circuit with topological basis group labels, where each group is a pair of gates that can be executed in parallel.

Inputs: circuit - a cirq.Circuit to be analyzed. This should be CZ + PhaseXZGate, otherwise no annotation will occur. weight_1q: float - the weight to assign to single-qubit gates. weight_2q: float - the weight to assign to two-qubit gates.

Returns: [0] - the cirq.Circuit with each gate annotated with topological similarity tags. [1] - a dictionary mapping each tag to its weight, where the key is the tag and the value is the weight.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/parallelize.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def auto_similarity(
    circuit: cirq.Circuit, weight_1q: float, weight_2q: float
) -> tuple[cirq.Circuit, dict[Hashable, float]]:
    """
    Automatically tag the circuit with topological basis group labels,
    where each group is a pair of gates that can be executed in parallel.

    Inputs:
    circuit - a cirq.Circuit to be analyzed. This should be CZ + PhaseXZGate, otherwise no annotation will occur.
    weight_1q: float - the weight to assign to single-qubit gates.
    weight_2q: float - the weight to assign to two-qubit gates.

    Returns:
    [0] - the cirq.Circuit with each gate annotated with topological similarity tags.
    [1] - a dictionary mapping each tag to its weight, where the key is the tag and the value is the weight.
    """
    flattened_circuit: list[GateOperation] = list(cirq.flatten_op_tree(circuit))
    weights = {}
    for i in range(len(flattened_circuit)):
        for j in range(i + 1, len(flattened_circuit)):
            op1 = flattened_circuit[i]
            op2 = flattened_circuit[j]
            if can_be_parallel(op1, op2):
                # Add tags to both operations
                tag = f"AUTO:{i}"
                flattened_circuit[i] = op1.with_tags(tag)
                flattened_circuit[j] = op2.with_tags(tag)
                if len(op1.qubits) == 1:
                    weights[tag] = weight_1q
                elif len(op1.qubits) == 2:
                    weights[tag] = weight_2q
                else:
                    raise RuntimeError("Unsupported gate type")
    return cirq.Circuit(flattened_circuit), weights

block_similarity 

block_similarity(
    circuit: Circuit, weight: float, block_id: int
) -> tuple[cirq.Circuit, dict[Hashable, float]]

Associate every gate in a circuit with a similarity group.

Inputs: circuit - a cirq.Circuit to be analyzed. weight: float - the weight to assign to each block of gates.

Returns: [0] - the cirq.Circuit with each gate annotated with topological similarity tags. [1] - a dictionary mapping each tag to its weight, where the key is the tag and the value is the weight.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/parallelize.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def block_similarity(
    circuit: cirq.Circuit, weight: float, block_id: int
) -> tuple[cirq.Circuit, dict[Hashable, float]]:
    """
    Associate every gate in a circuit with a similarity group.

    Inputs:
    circuit - a cirq.Circuit to be analyzed.
    weight: float - the weight to assign to each block of gates.

    Returns:
    [0] - the cirq.Circuit with each gate annotated with topological similarity tags.
    [1] - a dictionary mapping each tag to its weight, where the key is the tag and the value is the weight.
    """
    new_moments = []
    weights = {}
    tag = f"BLOCK:{block_id}"
    for moment in circuit.moments:
        new_moments.append([gate.with_tags(tag) for gate in moment.operations])
    weights[tag] = weight
    return cirq.Circuit(new_moments), weights

can_be_parallel 

can_be_parallel(
    op1: GateOperation,
    op2: GateOperation,
    tol: float = 1e-14,
) -> bool

Heuristic similarity function to determine if two operations are similar enough to be grouped together in parallel execution.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/parallelize.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def can_be_parallel(
    op1: cirq.GateOperation, op2: cirq.GateOperation, tol: float = 1e-14
) -> bool:
    """
    Heuristic similarity function to determine if two operations are similar enough
    to be grouped together in parallel execution.
    """
    are_disjoint = len(set(op1.qubits).intersection(op2.qubits)) == 0
    if not are_disjoint:
        return False

    # Check if both operations are CZ gates
    both_cz = op1.gate == cirq.CZ and op2.gate == cirq.CZ

    both_phased_xz = isinstance(op1.gate, cirq.PhasedXZGate) and isinstance(
        op2.gate, cirq.PhasedXZGate
    )
    equal_unitaries = cirq.equal_up_to_global_phase(
        cirq.unitary(op1.gate), cirq.unitary(op2.gate), atol=tol
    )

    return (both_phased_xz and equal_unitaries) or both_cz

colorize 

colorize(epochs: Iterable[list[Unique[GateOperation]]])

For each epoch, separate any 1q and 2q gates, and colorize the 2q gates so that they can be executed in parallel without conflicts. Args: epochs: list[list[Unique[cirq.GateOperation]]] - a list of epochs, where each epoch is a list of gates that can be executed in parallel.

Yields:

Type Description

list[cirq.GateOperation] - a list of lists of gates, where each inner list contains gates that can be executed in parallel.
Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/parallelize.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
def colorize(
    epochs: Iterable[list[Unique[cirq.GateOperation]]],
):
    """
    For each epoch, separate any 1q and 2q gates, and colorize the 2q gates
    so that they can be executed in parallel without conflicts.
    Args:
        epochs: list[list[Unique[cirq.GateOperation]]] - a list of epochs, where each
            epoch is a list of gates that can be executed in parallel.

    Yields:
        list[cirq.GateOperation] - a list of lists of gates, where each
            inner list contains gates that can be executed in parallel.

    """
    for epoch in epochs:
        oneq_gates = []
        twoq_gates = []
        for gate in epoch:
            if len(gate.val.qubits) == 1:
                oneq_gates.append(gate.val)
            elif len(gate.val.qubits) == 2:
                twoq_gates.append(gate.val)
            else:
                raise RuntimeError("Unsupported gate type")

        if len(oneq_gates) > 0:
            yield oneq_gates

        # twoq_gates2 = colorizer(twoq_gates)# Inlined.
        """
        Implements an edge coloring algorithm on a set of simultaneous 2q gates,
        so that they can be done in an ordered manner so that no to gates use
        the same qubit in the same layer.
        """
        graph = nx.Graph()
        for gate in twoq_gates:
            if len(gate.qubits) != 2 and gate.gate != cirq.CZ:
                raise RuntimeError("Unsupported gate type")
            graph.add_edge(gate.qubits[0], gate.qubits[1])
        linegraph = nx.line_graph(graph)

        best_colors: dict[tuple[cirq.Qid, cirq.Qid], int] = (
            nx.algorithms.coloring.greedy_color(linegraph, strategy="largest_first")
        )
        best_num_colors = len(set(best_colors.values()))

        for strategy in (
            #'random_sequential',
            "smallest_last",
            "independent_set",
            "connected_sequential_bfs",
            "connected_sequential_dfs",
            "saturation_largest_first",
        ):
            colors: dict[tuple[cirq.Qid, cirq.Qid], int] = (
                nx.algorithms.coloring.greedy_color(linegraph, strategy=strategy)
            )
            if (num_colors := len(set(colors.values()))) < best_num_colors:
                best_num_colors = num_colors
                best_colors = colors

        twoq_gates2 = (
            list(cirq.CZ(*k) for k, v in best_colors.items() if v == x)
            for x in set(best_colors.values())
        )
        # -- end colorizer --
        yield from twoq_gates2

generate_epochs 

generate_epochs(solution: dict[NodeType, float], tol=0.01)

Internal function to generate epochs from the solution of the linear program.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/parallelize.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def generate_epochs(
    solution: dict[NodeType, float],
    tol=1e-2,
):
    """
    Internal function to generate epochs from the solution of the linear program.
    """
    sorted_gates = sorted(solution.items(), key=lambda x: x[1])
    if len(sorted_gates) == 0:
        return iter([])

    gate, latest_time = sorted_gates[0]
    current_epoch = [gate]  # Start with the first gate
    for gate, time in sorted_gates[1:]:
        if time - latest_time < tol:
            current_epoch.append(gate)
        else:
            yield current_epoch
            current_epoch = [gate]

        latest_time = time

    yield current_epoch  # Yield the last epoch

moment_similarity 

moment_similarity(
    circuit: Circuit, weight: float
) -> tuple[cirq.Circuit, dict[Hashable, float]]

Associate every gate in each moment with a similarity group.

Inputs: circuit - a cirq.Circuit to be analyzed. weight: float - the weight to assign to each block of gates.

Returns: [0] - the cirq.Circuit with each gate annotated with topological similarity tags. [1] - a dictionary mapping each tag to its weight, where the key is the tag and the value is the weight.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/parallelize.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def moment_similarity(
    circuit: cirq.Circuit, weight: float
) -> tuple[cirq.Circuit, dict[Hashable, float]]:
    """
    Associate every gate in each moment with a similarity group.

    Inputs:
    circuit - a cirq.Circuit to be analyzed.
    weight: float - the weight to assign to each block of gates.

    Returns:
    [0] - the cirq.Circuit with each gate annotated with topological similarity tags.
    [1] - a dictionary mapping each tag to its weight, where the key is the tag and the value is the weight.
    """
    new_moments = []
    weights = {}

    for moment_index, moment in enumerate(circuit.moments):
        tag = f"MOMENT:{moment_index}"
        new_moments.append([gate.with_tags(tag) for gate in moment.operations])
        weights[tag] = weight
    return cirq.Circuit(new_moments), weights

no_similarity 

no_similarity(circuit: Circuit) -> cirq.Circuit

Removes all tags from the circuit

Inputs: circuit: cirq.Circuit - the circuit to remove tags from.

Returns: [0] - cirq.Circuit - the circuit with all tags removed.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/parallelize.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def no_similarity(circuit: cirq.Circuit) -> cirq.Circuit:
    """
    Removes all tags from the circuit

    Inputs:
    circuit: cirq.Circuit - the circuit to remove tags from.

    Returns:
    [0] - cirq.Circuit - the circuit with all tags removed.
    """
    new_moments = []
    for moment in circuit.moments:
        new_moments.append([gate.untagged for gate in moment.operations])
    return cirq.Circuit(new_moments)

parallelize 

parallelize(
    circuit: Circuit,
    hyperparameters: dict[str, float] | None = None,
    auto_tag: bool = True,
) -> cirq.Circuit

Use linear programming to reorder a circuit so that it may be optimally be run in parallel. This is done using a DAG representation, as well as a heuristic similarity function to group parallelizable gates together.

Extra topological information (similarity) can be used by tagging each gate with the topological basis groups that it belongs to, for example

circuit.append(cirq.H(qubits[0]).with_tags(1,2,3,4)) represents that this gate is part of the topological basis groups 1,2,3, and 4.

Inputs

circuit: cirq.Circuit - the static circuit to be optimized hyperparameters: dict[str, float] - hyperparameters for the optimization - "linear": float (0.01) - the linear cost of each gate - "1q": float (1.0) - the quadratic cost of 1q gates - "2q": float (2.0) - the quadratic cost of 2q gates - "tags": float (0.5) - the default weight of the topological basis.

Returns: cirq.Circuit - the optimized circuit, where each moment is as parallel as possible. it is also broken into native CZ gate set of {CZ, PhXZ}

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/parallelize.py 
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
def parallelize(
    circuit: cirq.Circuit,
    hyperparameters: dict[str, float] | None = None,
    auto_tag: bool = True,
) -> cirq.Circuit:
    """
    Use linear programming to reorder a circuit so that it may be optimally be
    run in parallel. This is done using a DAG representation, as well as a heuristic
    similarity function to group parallelizable gates together.

    Extra topological information (similarity) can be used by tagging each gate with
    the topological basis groups that it belongs to, for example
    > circuit.append(cirq.H(qubits[0]).with_tags(1,2,3,4))
    represents that this gate is part of the topological basis groups 1,2,3, and 4.

    Inputs:
        circuit: cirq.Circuit - the static circuit to be optimized
        hyperparameters: dict[str, float] - hyperparameters for the optimization
            - "linear": float (0.01) - the linear cost of each gate
            - "1q": float (1.0)  - the quadratic cost of 1q gates
            - "2q": float (2.0)  - the quadratic cost of 2q gates
            - "tags": float (0.5) - the default weight of the topological basis.
    Returns:
        cirq.Circuit - the optimized circuit, where each moment is as parallel as possible.
          it is also broken into native CZ gate set of {CZ, PhXZ}
    """
    hyperparameters = _get_hyperparameters(hyperparameters)

    # Transpile the circuit to a native CZ gate set.
    transpiled_circuit = transpile(circuit)
    if auto_tag:
        # Annotate the circuit with topological information
        # to improve parallelization
        transpiled_circuit, group_weights = auto_similarity(
            transpiled_circuit,
            weight_1q=hyperparameters.get("1q", 1.0),
            weight_2q=hyperparameters.get("2q", 1.0),
        )
    else:
        group_weights = {}
    epochs = colorize(
        generate_epochs(
            solve_epochs(
                directed=to_dag_circuit(transpiled_circuit),
                group_weights=group_weights,
                hyperparameters=hyperparameters,
            )
        )
    )
    # Convert the epochs to a cirq circuit.
    moments = map(cirq.Moment, epochs)
    return cirq.Circuit(moments)

solve_epochs 

solve_epochs(
    directed: DiGraph,
    group_weights: dict[Hashable, float],
    hyperparameters: dict[str, float] | None = None,
) -> dict[Unique[cirq.GateOperation], float]

Internal function to solve the epochs using linear programming.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/parallelize.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def solve_epochs(
    directed: nx.DiGraph,
    group_weights: dict[Hashable, float],
    hyperparameters: dict[str, float] | None = None,
) -> dict[Unique[cirq.GateOperation], float]:
    """
    Internal function to solve the epochs using linear programming.
    """

    hyperparameters = _get_hyperparameters(hyperparameters)

    basis = {node: Variable() for node in directed.nodes}

    if len(basis) == 0:
        return {}

    # ---
    # Turn into a linear program to solve
    # ---
    lp = LPProblem()

    # All timesteps must be positive
    for node in directed.nodes:
        lp.add_gez(1.0 * basis[node])

    # Add ordering constraints
    for edge in directed.edges:
        lp.add_gez(basis[edge[1]] - basis[edge[0]] - 1.0)

    all_variables = list(basis.values())
    # Add linear objective: minimize the total time
    objective = hyperparameters["linear"] * sum(all_variables[1:], all_variables[0])

    default_weight = hyperparameters["tags"]
    lp.add_linear(objective)
    # Add ABS objective: similarity wants to go together.
    for node1, node2 in combinations(directed.nodes, 2):
        # Topological (user) similarity:
        inter = set(node1.val.tags).intersection(set(node2.val.tags))
        if len(inter) > 0:
            weight = sum([group_weights.get(key, default_weight) for key in inter])
            if weight > 0:
                lp.add_abs((basis[node1] - basis[node2]) * weight)
            elif weight < 0:
                raise RuntimeError("Weights must be positive")

    solution = lp.solve()
    return {node: solution[basis[node]] for node in directed.nodes}

to_dag_circuit 

to_dag_circuit(
    circuit: Circuit, can_reorder=None
) -> nx.DiGraph

Convert a cirq.Circuit to a directed acyclic graph (DAG) representation. This is useful for analyzing the circuit structure and dependencies.

Parameters:

Name Type Description Default
circuit Circuit

cirq.Circuit - the circuit to convert.

 required
can_reorder

function - a function that checks if two operations can be reordered.

 None

Returns: [0] - nx.DiGraph - the directed acyclic graph representation of the circuit.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/parallelize.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def to_dag_circuit(circuit: cirq.Circuit, can_reorder=None) -> nx.DiGraph:
    """
    Convert a cirq.Circuit to a directed acyclic graph (DAG) representation.
    This is useful for analyzing the circuit structure and dependencies.

    Args:
        circuit: cirq.Circuit - the circuit to convert.
        can_reorder: function - a function that checks if two operations can be reordered.

    Returns:
    [0] - nx.DiGraph - the directed acyclic graph representation of the circuit.
    """

    def reorder_check(
        op1, op2
    ):  # can reorder iff both are CZ, or intersection is empty
        if op1.gate == cirq.CZ and op2.gate == cirq.CZ:
            return True
        else:
            return len(set(op1.qubits).intersection(op2.qubits)) == 0

    # Turn into DAG
    directed = CircuitDag.from_circuit(
        circuit, can_reorder=reorder_check if can_reorder is None else can_reorder
    )
    return nx.transitive_reduction(directed)

transpile 

transpile(circuit: Circuit) -> cirq.Circuit

Transpile a circuit to a native CZ gate set of {CZ, PhXZ}.

Source code in .venv/lib/python3.12/site-packages/bloqade/cirq_utils/parallelize.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def transpile(circuit: cirq.Circuit) -> cirq.Circuit:
    """
    Transpile a circuit to a native CZ gate set of {CZ, PhXZ}.
    """
    # Convert to CZ target gate set.
    circuit2 = cirq.optimize_for_target_gateset(circuit, gateset=cirq.CZTargetGateset())
    circuit2 = cirq.drop_empty_moments(circuit2)

    missing_qubits = circuit.all_qubits() - circuit2.all_qubits()

    for qubit in missing_qubits:
        circuit2.append(
            cirq.PhasedXZGate(x_exponent=0, z_exponent=0, axis_phase_exponent=0).on(
                qubit
            )
        )

    return circuit2