Source code for kosmos.protocols.circuit_execution_protocol

from qiskit import QuantumCircuit

from kosmos.circuit_runner.qiskit_runner import AerSimulatorRunner, QiskitRunner
from kosmos.protocols.config.protocol import CircuitExecutionProtocolConfig
from kosmos.protocols.protocol import Protocol
from kosmos.protocols.protocol_result import CircuitExecutionProtocolResult
from kosmos.protocols.status import ProtocolStatus
from kosmos.topology.net import Network




class CircuitExecutionProtocol(Protocol):
    """Protocol for executing a circuit partition on a quantum node.

    Attributes:
        circuit_runner (QiskitRunner): The circuit runner to execute the circuit.
        circuit (QuantumCircuit): The quantum circuit to execute.

    """

    def __init__(
        self,
        config: CircuitExecutionProtocolConfig,
        network: Network,
        circuit_runner: QiskitRunner,
        circuit: QuantumCircuit,
    ) -> None:
        """Initialize the circuit execution protocol.

        Args:
            config (CircuitExecutionProtocolConfig): Circuit execution config.
            network (Network): The network topology.
            circuit_runner (QiskitRunner): The circuit runner to execute the circuit.
            circuit (QuantumCircuit): The quantum circuit to execute.

        """
        super().__init__(config, network)
        self.config = config
        self.circuit_runner = circuit_runner
        self.circuit = circuit



    def execute(self) -> CircuitExecutionProtocolResult:
        """Execute the quantum circuit.

        Returns:
            CircuitExecutionProtocolResult: Result of the circuit execution.

        """
        self.status = ProtocolStatus.RUNNING
        density_matrix = None

        if isinstance(self.circuit_runner, AerSimulatorRunner):
            density_matrix, _ = self.circuit_runner.density_matrix(self.circuit)

        execution_time = (
            self.circuit.depth() * (self.config.local_gate_time * 2)
            + self.circuit.num_qubits * self.config.measurement_time
        )

        self.status = ProtocolStatus.SUCCESS

        return CircuitExecutionProtocolResult(
            status=self.status,
            execution_time=execution_time,
            density_matrix=density_matrix,
        )