def serialize_circuit()

in tensorflow_quantum/core/serialize/serializer.py [0:0]

• 46 lines of code
• 19 McCabe index (conditional complexity)

def serialize_circuit(circuit_inp):
    """Returns a `cirq.Program` proto representing the `cirq.Circuit`.

    Note that the circuit must use gates valid in the tfq_gate_set.
    Currently we only support scalar multiplication of symbols and
    no other more complex arithmetic expressions. This means
    we can support things like X**(3*alpha), and Rx(alpha). Because
    we use the `cirq.Program` proto, we only support `cirq.GridQubit`
    and `cirq.LineQubit` instances during serialization of circuits.

    Note: once serialized terminal measurements are removed.

    Args:
        circuit_inp: A `cirq.Circuit`.

    Returns:
        A `tfq.proto.Program` proto.
    """
    circuit = copy.deepcopy(circuit_inp)
    if not isinstance(circuit, cirq.Circuit):
        raise TypeError("serialize requires cirq.Circuit objects."
                        " Given: " + str(type(circuit)))

    # This code is intentionally written to avoid using cirq functions
    # as this get analyzed by tensorflow-autograph.

    # Gives a map from moment index to measure qubits in moment
    measured_moments = dict()

    # Tracks qubits that have been measured already.
    all_measured_qubits = set()
    for i, moment in enumerate(circuit.moments):
        measured_qubits = set()
        for op in moment:
            for qubit in op.qubits:
                if not isinstance(qubit, (cirq.GridQubit, cirq.LineQubit)):
                    raise ValueError(
                        "Attempted to serialize circuit that don't use "
                        "only cirq.GridQubits or cirq.LineQubits.")

            if isinstance(op.gate, cirq.MeasurementGate):
                for qubit in op.qubits:
                    if qubit in all_measured_qubits:
                        raise ValueError("Serialization of circuit failed. "
                                         "Circuits with non-terminal "
                                         "measurement operations are not "
                                         "supported.")
                    measured_qubits.add(qubit)
                    all_measured_qubits.add(qubit)

        if len(measured_qubits) > 0:
            measured_moments[i] = measured_qubits

    # Remove terminal measurements.
    for moment_ind in measured_moments:
        old_moment = circuit[moment_ind]
        measured_qubits = measured_moments[moment_ind]
        new_moment = cirq.Moment(
            filter(lambda x: not any(y in measured_qubits for y in x.qubits),
                   old_moment.operations))
        circuit[moment_ind] = new_moment

    # Demote cirq.controlled_operations (controlled gates) to their sub_gate
    # types with _tfq_control_qubits and _tfq_control_values fields so that
    # the gates can still get picked up by the serializer. There would be no way
    # to discern controlledgates from one another otherwise. This
    # "momentary demotion" occurs with the help of the DelayedAssignmentGate.
    for i, moment in enumerate(circuit):
        controlled_ops = [
            op for op in moment if isinstance(op, cirq.ControlledOperation)
        ]
        new_ops = dict()
        for op in controlled_ops:
            tfq_compatible = op.sub_operation
            tfq_compatible._tfq_control_qubits = op.controls
            tfq_compatible._tfq_control_values = op.control_values
            new_ops[op.qubits] = tfq_compatible

        circuit[i] = cirq.Moment(
            new_ops[op.qubits] if op.qubits in new_ops else op for op in moment)

    return SERIALIZER.serialize(circuit)